class AutopushConfig(object):
    """Main Autopush Settings Object"""

    debug = attrib(default=False)  # type: bool

    fernet = attrib(init=False)  # type: MultiFernet
    _crypto_key = attrib(
        converter=_init_crypto_key, default=None)  # type: List[str]

    bear_hash_key = attrib(default=Factory(list))  # type: List[str]
    human_logs = attrib(default=True)  # type: bool

    hostname = attrib(default=None)  # type: Optional[str]
    port = attrib(default=None)  # type: Optional[int]
    _resolve_hostname = attrib(default=False)  # type: bool

    router_scheme = attrib(default=None)  # type: Optional[str]
    router_hostname = attrib(default=None)  # type: Optional[str]
    router_port = attrib(default=None)  # type: Optional[int]

    endpoint_scheme = attrib(default=None)  # type: Optional[str]
    endpoint_hostname = attrib(default=None)  # type: Optional[str]
    endpoint_port = attrib(default=None)  # type: Optional[int]

    proxy_protocol_port = attrib(default=None)  # type: Optional[int]
    memusage_port = attrib(default=None)  # type: Optional[int]

    statsd_host = attrib(default="localhost")  # type: str
    statsd_port = attrib(default=8125)  # type: int
    megaphone_api_url = attrib(default=None)  # type: Optional[str]
    megaphone_api_token = attrib(default=None)  # type: Optional[str]
    megaphone_poll_interval = attrib(default=30)  # type: int

    datadog_api_key = attrib(default=None)  # type: Optional[str]
    datadog_app_key = attrib(default=None)  # type: Optional[str]
    datadog_flush_interval = attrib(default=None)  # type: Optional[int]

    router_table = _nested(
        DDBTableConfig,
        default=dict(tablename="router")
    )  # type: DDBTableConfig
    message_table = _nested(
        DDBTableConfig,
        default=dict(tablename="message")
    )  # type: DDBTableConfig

    preflight_uaid = attrib(
        default="deadbeef00000000deadbeef00000000")  # type: str

    ssl = _nested(SSLConfig, default=Factory(SSLConfig))  # type: SSLConfig
    router_ssl = _nested(
        SSLConfig, default=Factory(SSLConfig))  # type: SSLConfig
    client_certs = attrib(default=None)  # type: Optional[Dict[str, str]]

    router_url = attrib(init=False)  # type: str
    endpoint_url = attrib(init=False)  # type: str
    ws_url = attrib(init=False)  # type: str

    router_conf = attrib(default=Factory(dict))  # type: JSONDict

    # twisted Agent's connectTimeout
    connect_timeout = attrib(default=0.5)  # type: float
    max_data = attrib(default=4096)  # type: int
    env = attrib(default='development')  # type: str
    ami_id = attrib(default=None)  # type: Optional[str]
    cors = attrib(default=False)  # type: bool

    hello_timeout = attrib(default=0)  # type: int
    # Force timeout in idle seconds
    msg_limit = attrib(default=100)  # type: int
    auto_ping_interval = attrib(default=None)  # type: Optional[int]
    auto_ping_timeout = attrib(default=None)  # type: Optional[int]
    max_connections = attrib(default=None)  # type: Optional[int]
    close_handshake_timeout = attrib(default=None)  # type: Optional[int]

    # Generate messages per legacy rules, only used for testing to
    # generate legacy data.
    _notification_legacy = attrib(default=False)  # type: bool

    # Use the cryptography library
    use_cryptography = attrib(default=False)  # type: bool

    # Strict-Transport-Security max age (Default 1 year in secs)
    sts_max_age = attrib(default=31536000)  # type: int

    # Don't cache ssl.wrap_socket's SSLContexts
    no_sslcontext_cache = attrib(default=False)  # type: bool

    # DynamoDB endpoint override
    aws_ddb_endpoint = attrib(default=None)  # type: str

    allow_table_rotation = attrib(default=True)  # type: bool

    def __attrs_post_init__(self):
        """Initialize the Settings object"""
        # Setup hosts/ports/urls
        if not self.hostname:
            self.hostname = socket.gethostname()
        if self._resolve_hostname:
            self.hostname = resolve_ip(self.hostname)

        if not self.endpoint_hostname:
            self.endpoint_hostname = self.hostname
        if not self.router_hostname:
            self.router_hostname = self.hostname

        self.router_url = canonical_url(
            self.router_scheme or 'http',
            self.router_hostname,
            self.router_port
        )
        self.endpoint_url = canonical_url(
            self.endpoint_scheme or 'http',
            self.endpoint_hostname,
            self.endpoint_port
        )
        # not accurate under autoendpoint (like router_url)
        self.ws_url = "{}://{}:{}/".format(
            'wss' if self.ssl.key else 'ws',
            self.hostname,
            self.port
        )

        self.fernet = MultiFernet([Fernet(key) for key in self._crypto_key])

    @property
    def enable_tls_auth(self):
        """Whether TLS authentication w/ client certs is enabled"""
        return self.client_certs is not None

    @classmethod
    def from_argparse(cls, ns, **kwargs):
        # type: (Namespace, **Any) -> AutopushConfig
        """Create an instance from argparse/additional kwargs"""
        router_conf = {}
        if ns.key_hash:
            db.key_hash = ns.key_hash
        if ns.apns_creds:
            # if you have the critical elements for each external
            # router, create it
            try:
                router_conf["apns"] = json.loads(ns.apns_creds)
            except (ValueError, TypeError):
                raise InvalidConfig(
                    "Invalid JSON specified for APNS config options")
        if ns.senderid_list:
            # Create a common gcmclient
            try:
                sender_ids = json.loads(ns.senderid_list)
            except (ValueError, TypeError):
                raise InvalidConfig("Invalid JSON specified for senderid_list")
            try:
                # This is an init check to verify that things are
                # configured correctly. Otherwise errors may creep in
                # later that go unaccounted.
                sender_ids[sender_ids.keys()[0]]
            except (IndexError, TypeError):
                raise InvalidConfig("No GCM SenderIDs specified or found.")
            router_conf["gcm"] = {"ttl": ns.gcm_ttl,
                                  "dryrun": ns.gcm_dryrun,
                                  "max_data": ns.max_data,
                                  "collapsekey": ns.gcm_collapsekey,
                                  "senderIDs": sender_ids,
                                  "endpoint": ns.gcm_endpoint}
        client_certs = None
        # endpoint only
        if getattr(ns, 'client_certs', None):
            try:
                client_certs_arg = json.loads(ns.client_certs)
            except (ValueError, TypeError):
                raise InvalidConfig("Invalid JSON specified for client_certs")
            if client_certs_arg:
                if not ns.ssl_key:
                    raise InvalidConfig("client_certs specified without SSL "
                                        "enabled (no ssl_key specified)")
                client_certs = {}
                for name, sigs in client_certs_arg.iteritems():
                    if not isinstance(sigs, list):
                        raise InvalidConfig(
                            "Invalid JSON specified for client_certs")
                    for sig in sigs:
                        sig = sig.upper()
                        if (not name or not CLIENT_SHA256_RE.match(sig) or
                                sig in client_certs):
                            raise InvalidConfig(
                                "Invalid client_certs argument")
                        client_certs[sig] = name

        if ns.fcm_creds:
            try:
                router_conf["fcm"] = {
                    "version": ns.fcm_version,
                    "ttl": ns.fcm_ttl,
                    "dryrun": ns.fcm_dryrun,
                    "max_data": ns.max_data,
                    "collapsekey": ns.fcm_collapsekey,
                    "creds": json.loads(ns.fcm_creds)
                }
                if not router_conf["fcm"]["creds"]:
                    raise InvalidConfig(
                        "Empty credentials for FCM config options"
                    )
                for creds in router_conf["fcm"]["creds"].values():
                    if "auth" not in creds:
                        raise InvalidConfig(
                            "Missing auth for FCM config options"
                        )
            except (ValueError, TypeError):
                raise InvalidConfig(
                    "Invalid JSON specified for FCM config options"
                )

        if ns.adm_creds:
            # Create a common admclient
            try:
                router_conf["adm"] = json.loads(ns.adm_creds)
            except (ValueError, TypeError):
                raise InvalidConfig(
                    "Invalid JSON specified for ADM config options")

        ami_id = None
        # Not a fan of double negatives, but this makes more
        # understandable args
        if not ns.no_aws:
            ami_id = get_amid() or "Unknown"

        allow_table_rotation = not ns.no_table_rotation
        return cls(
            crypto_key=ns.crypto_key,
            datadog_api_key=ns.datadog_api_key,
            datadog_app_key=ns.datadog_app_key,
            datadog_flush_interval=ns.datadog_flush_interval,
            hostname=ns.hostname,
            statsd_host=ns.statsd_host,
            statsd_port=ns.statsd_port,
            router_conf=router_conf,
            resolve_hostname=ns.resolve_hostname,
            ami_id=ami_id,
            client_certs=client_certs,
            msg_limit=ns.msg_limit,
            connect_timeout=ns.connection_timeout,
            memusage_port=ns.memusage_port,
            use_cryptography=ns.use_cryptography,
            no_sslcontext_cache=ns._no_sslcontext_cache,
            router_table=dict(
                tablename=ns.router_tablename,
                read_throughput=ns.router_read_throughput,
                write_throughput=ns.router_write_throughput
            ),
            message_table=dict(
                tablename=ns.message_tablename,
                read_throughput=ns.message_read_throughput,
                write_throughput=ns.message_write_throughput
            ),
            ssl=dict(
                key=ns.ssl_key,
                cert=ns.ssl_cert,
                dh_param=ns.ssl_dh_param
            ),
            sts_max_age=ns.sts_max_age,
            allow_table_rotation=allow_table_rotation,
            **kwargs
        )

    def make_endpoint(self, uaid, chid, key=None):
        """Create an v1 or v2 WebPush endpoint from the identifiers.

        Both endpoints use bytes instead of hex to reduce ID length.
        v1 is the uaid + chid
        v2 is the uaid + chid + sha256(key).bytes

        :param uaid: User Agent Identifier
        :param chid: Channel or Subscription ID
        :param key: Optional Base64 URL-encoded application server key
        :returns: Push endpoint

        """
        root = self.endpoint_url + '/wpush/'
        base = (uaid.replace('-', '').decode("hex") +
                chid.replace('-', '').decode("hex"))

        if key is None:
            return root + 'v1/' + self.fernet.encrypt(base).strip('=')

        raw_key = base64url_decode(key.encode('utf8'))
        ep = self.fernet.encrypt(base + sha256(raw_key).digest()).strip('=')
        return root + 'v2/' + ep

    def parse_endpoint(self, metrics, token, version="v1", ckey_header=None,
                       auth_header=None):
        """Parse an endpoint into component elements of UAID, CHID and optional
        key hash if v2

        :param token: The obscured subscription data.
        :param version: This is the API version of the token.
        :param ckey_header: the Crypto-Key header bearing the public key
            (from Crypto-Key: p256ecdsa=)
        :param auth_header: The Authorization header bearing the VAPID info

        :raises ValueError: In the case of a malformed endpoint.

        :returns: a dict containing (uaid=UAID, chid=CHID, public_key=KEY)

        """
        token = self.fernet.decrypt(repad(token).encode('utf8'))
        public_key = None
        if ckey_header:
            try:
                crypto_key = CryptoKey(ckey_header)
            except CryptoKeyException:
                raise InvalidTokenException("Invalid key data")
            public_key = crypto_key.get_label('p256ecdsa')
        if auth_header:
            vapid_auth = parse_auth_header(auth_header)
            if not vapid_auth:
                raise VapidAuthException("Invalid Auth token")
            metrics.increment("notification.auth",
                              tags="vapid:{version},scheme:{scheme}".format(
                                  **vapid_auth
                              ).split(","))
            # pull the public key from the VAPID auth header if needed
            try:
                if vapid_auth['version'] != 1:
                    public_key = vapid_auth['k']
            except KeyError:
                raise VapidAuthException("Missing Public Key")
        if version == 'v1' and len(token) != 32:
            raise InvalidTokenException("Corrupted push token")
        if version == 'v2':
            if not auth_header:
                raise VapidAuthException("Missing Authorization Header")
            if len(token) != 64:
                raise InvalidTokenException("Corrupted push token")
            if not public_key:
                raise VapidAuthException("Invalid key data")
            try:
                decoded_key = base64url_decode(public_key)
            except TypeError:
                raise VapidAuthException("Invalid key data")
            if not constant_time.bytes_eq(sha256(decoded_key).digest(),
                                          token[32:]):
                raise VapidAuthException("Key mismatch")
        return dict(uaid=token[:16].encode('hex'),
                    chid=token[16:32].encode('hex'),
                    version=version,
                    public_key=public_key) class DatabaseManager(object):
    """Provides database access"""

    _router_conf = attrib()   # type: DDBTableConfig
    _message_conf = attrib()  # type: DDBTableConfig
    metrics = attrib()        # type: IMetrics
    resource = attrib()       # type: DynamoDBResource

    router = attrib(default=None)                   # type: Optional[Router]
    message_tables = attrib(default=Factory(list))  # type: List[str]
    current_msg_month = attrib(init=False)          # type: Optional[str]
    current_month = attrib(init=False)              # type: Optional[int]
    _message = attrib(default=None)                 # type: Optional[Message]
    allow_table_rotation = attrib(default=True)     # type: Optional[bool]
    # for testing:

    def __attrs_post_init__(self):
        """Initialize sane defaults"""
        if self.allow_table_rotation:
            today = datetime.date.today()
            self.current_month = today.month
            self.current_msg_month = make_rotating_tablename(
                self._message_conf.tablename,
                date=today
            )
        else:
            # fetch out the last message table as the "current_msg_month"
            # Message may still init to this table if it recv's None, but
            # this makes the value explicit.
            resource = self.resource
            self.current_msg_month = resource.get_latest_message_tablename(
                prefix=self._message_conf.tablename
            )

        if not self.resource:
            self.resource = DynamoDBResource()

    @classmethod
    def from_config(cls,
                    conf,           # type: AutopushConfig
                    resource=None,  # type: Optional[DynamoDBResource]
                    **kwargs        # type: Any
                    ):
        # type: (...) -> DatabaseManager
        """Create a DatabaseManager from the given config"""
        metrics = autopush.metrics.from_config(conf)
        if not resource:
            resource = DynamoDBResource()
        return cls(
            router_conf=conf.router_table,
            message_conf=conf.message_table,
            metrics=metrics,
            resource=resource,
            allow_table_rotation=conf.allow_table_rotation,
            **kwargs
        )

    def setup(self, preflight_uaid):
        # type: (str) -> None
        """Setup metrics, message tables and perform preflight_check"""
        self.metrics.start()
        self.setup_tables()
        preflight_check(self.message, self.router, preflight_uaid)

    def setup_tables(self):
        """Lookup or create the database tables"""
        self.router = Router(
            conf=self._router_conf,
            metrics=self.metrics,
            resource=self.resource)
        # Used to determine whether a connection is out of date with current
        # db objects. There are three noteworty cases:
        # 1 "Last Month" the table requires a rollover.
        # 2 "This Month" the most common case.
        # 3 "Next Month" where the system will soon be rolling over, but with
        #   timing, some nodes may roll over sooner. Ensuring the next month's
        #   table is present before the switchover is the main reason for this,
        #   just in case some nodes do switch sooner.
        self.create_initial_message_tables()
        self._message = Message(self.current_msg_month,
                                boto_resource=self.resource)

    @property
    def message(self):
        # type: () -> Message
        """Property that access the current message table"""
        if not self._message or isinstance(self._message, Attribute):
            self._message = self.message_table(self.current_msg_month)
        return self._message

    def message_table(self, tablename):
        return Message(tablename, boto_resource=self.resource)

    def _tomorrow(self):
        # type: () -> datetime.date
        return datetime.date.today() + datetime.timedelta(days=1)

    def create_initial_message_tables(self):
        """Initializes a dict of the initial rotating messages tables.

        An entry for last months table, an entry for this months table,
        an entry for tomorrow, if tomorrow is a new month.

        """
        if not self.allow_table_rotation:
            tablenames = self.resource.get_latest_message_tablenames(
                prefix=self._message_conf.tablename,
                previous=3
            )
            # Create the most recent table if it's not there.
            tablename = tablenames[-1]
            if not table_exists(tablename,
                                boto_resource=self.resource):
                create_message_table(
                    tablename=tablename,
                    read_throughput=self._message_conf.read_throughput,
                    write_throughput=self._message_conf.write_throughput,
                    boto_resource=self.resource
                )
            self.message_tables.extend(tablenames)
            return

        mconf = self._message_conf
        today = datetime.date.today()
        last_month = get_rotating_message_tablename(
            prefix=mconf.tablename,
            delta=-1,
            message_read_throughput=mconf.read_throughput,
            message_write_throughput=mconf.write_throughput,
            boto_resource=self.resource,
        )
        this_month = get_rotating_message_tablename(
            prefix=mconf.tablename,
            message_read_throughput=mconf.read_throughput,
            message_write_throughput=mconf.write_throughput,
            boto_resource=self.resource,
        )
        self.current_month = today.month
        self.current_msg_month = this_month
        self.message_tables = [last_month, this_month]
        if self._tomorrow().month != today.month:
            next_month = get_rotating_message_tablename(
                prefix=mconf.tablename,
                delta=1,
                message_read_throughput=mconf.read_throughput,
                message_write_throughput=mconf.write_throughput,
                boto_resource=self.resource,
            )
            self.message_tables.append(next_month)

    @inlineCallbacks
    def update_rotating_tables(self):
        # type: () -> Generator
        """This method is intended to be tasked to run periodically off the
        twisted event hub to rotate tables.

        When today is a new month from yesterday, then we swap out all the
        table objects on the settings object.

        """
        if not self.allow_table_rotation:
            returnValue(False)
        mconf = self._message_conf
        today = datetime.date.today()
        tomorrow = self._tomorrow()
        if ((tomorrow.month != today.month) and
                sorted(self.message_tables)[-1] != tomorrow.month):
            next_month = yield deferToThread(
                get_rotating_message_tablename,
                prefix=mconf.tablename,
                delta=0,
                date=tomorrow,
                message_read_throughput=mconf.read_throughput,
                message_write_throughput=mconf.write_throughput,
                boto_resource=self.resource
            )
            self.message_tables.append(next_month)
        if today.month == self.current_month:
            # No change in month, we're fine.
            returnValue(False)

        # Get tables for the new month, and verify they exist before we
        # try to switch over
        message_table = yield deferToThread(
            get_rotating_message_tablename,
            prefix=mconf.tablename,
            message_read_throughput=mconf.read_throughput,
            message_write_throughput=mconf.write_throughput,
            boto_resource=self.resource,
        )

        # Both tables found, safe to switch-over
        self.current_month = today.month
        self.current_msg_month = message_table
        returnValue(True) class InternalRouterHTTPFactory(BaseHTTPFactory):

    ap_handlers = (
        (r"/push/([^\/]+)", RouterHandler),
        (r"/notif/([^\/]+)(?:/(\d+))?", NotificationHandler),
    )

    def __init__(self,
                 conf,         # type: AutopushConfig
                 db,           # type: DatabaseManager
                 clients,      # type: Dict[str, PushServerProtocol]
                 **kwargs):
        # type: (...) -> None
        BaseHTTPFactory.__init__(self, conf, db, **kwargs)
        self.clients = clients

    @property
    def _hostname(self):
        return self.conf.router_hostname

    def ssl_cf(self):
        # type: () -> Optional[AutopushSSLContextFactory]
        """Build our SSL Factory (if configured).

        Configured from the router_ssl_key/cert and ssl_dh_param
        values.

        """
        return self.conf.router_ssl.cf()

    @classmethod
    def _for_handler(cls, conf, db, clients=None, **kwargs):
        if clients is None:
            clients = {}
        return cls(conf, db=db, clients=clients, **kwargs) class SinkMetrics(IMetrics):
    """Exists to ignore metrics when metrics are not active"""
    def increment(self, name, count=1, **kwargs):
        pass

    def gauge(self, name, count, **kwargs):
        pass

    def timing(self, name, duration, **kwargs):
        pass class WebPushNotification(object):
    """WebPush Notification

    This object centralizes all logic involving the addressing of a single
    WebPush Notification.

    message_id serves a complex purpose. It's returned as the Location header
    value so that an application server may delete the message. It's used as
    part of the non-versioned sort-key. Due to this, its an encrypted value
    that contains the necessary information to derive the location of this
    precise message in the appropriate message table.

    """
    uaid = attrib()  # type: uuid.UUID
    channel_id = attrib()  # type: uuid.UUID
    ttl = attrib()  # type: int
    data = attrib(default=None)  # type: Optional[str]
    headers = attrib(default=None)  # type: Optional[Dict[str, str]]
    timestamp = attrib(default=Factory(lambda: int(time.time())))  # type: int
    sortkey_timestamp = attrib(default=None)  # type: Optional[int]
    topic = attrib(default=None)  # type: Optional[str]
    source = attrib(default="Direct")  # type: Optional[str]

    message_id = attrib(default=None)  # type: str

    # Not an alias for message_id, for backwards compat and cases where an old
    # message with any update_id should be removed.
    update_id = attrib(default=None)  # type: str

    # Whether this notification should follow legacy non-topic rules
    legacy = attrib(default=False)  # type: bool

    def generate_message_id(self, fernet):
        # type: (Fernet) -> str
        """Generate a message-id suitable for accessing the message

        For topic messages, a sort_key version of 01 is used, and the topic
        is included for reference:

            Encrypted('01' : uaid.hex : channel_id.hex : topic)

        For non-topic messages, a sort_key version of 02 is used:

            Encrypted('02' : uaid.hex : channel_id.hex : timestamp)

        For legacy non-topic messages, no sort_key version was used and the
        message-id was:

            Encrypted('m' : uaid.hex : channel_id.hex)

        This is a blocking call.

        """
        if self.topic:
            msg_key = ":".join(["01", self.uaid.hex, self.channel_id.hex,
                                self.topic])
        elif self.legacy:
            msg_key = ":".join(["m", self.uaid.hex, self.channel_id.hex])
        else:
            self.sortkey_timestamp = self.sortkey_timestamp or ns_time()
            msg_key = ":".join(["02", self.uaid.hex, self.channel_id.hex,
                                str(self.sortkey_timestamp)])
        self.message_id = fernet.encrypt(msg_key.encode('utf8'))
        self.update_id = self.message_id
        return self.message_id

    @staticmethod
    def parse_decrypted_message_id(decrypted_token):
        # type: (str) -> Dict[str, Any]
        """Parses a decrypted message-id into component parts"""
        topic = None
        sortkey_timestamp = None
        if decrypted_token.startswith("01:"):
            info = decrypted_token.split(":")
            if len(info) != 4:
                raise InvalidTokenException("Incorrect number of token parts.")
            api_ver, uaid, chid, topic = info
        elif decrypted_token.startswith("02:"):
            info = decrypted_token.split(":")
            if len(info) != 4:
                raise InvalidTokenException("Incorrect number of token parts.")
            api_ver, uaid, chid, raw_sortkey = info
            sortkey_timestamp = int(raw_sortkey)
        else:
            info = decrypted_token.split(":")
            if len(info) != 3:
                raise InvalidTokenException("Incorrect number of token parts.")
            kind, uaid, chid = decrypted_token.split(":")
            if kind != "m":
                raise InvalidTokenException("Incorrect token kind.")
        return dict(
            uaid=uaid,
            chid=chid,
            topic=topic,
            sortkey_timestamp=sortkey_timestamp,
        )

    def cleanup_headers(self):
        # type: () -> None
        """Sanitize the headers for this notification

        This only needs to be run when creating a notification from passed
        in application server headers.

        """
        headers = self.headers
        # Strip crypto/encryption headers down
        for hdr in ["crypto-key", "encryption"]:
            if hdr in headers:
                head = headers[hdr].replace('"', '')
                head = STRIP_PADDING.sub("", head)
                head.decode('ascii')
                headers[hdr] = head

        # content-encoding header may already be stored as "encoding",
        # this is a failover to ensure that the proper value is pulled in.
        # Webpush expects the value of "encoding".
        data = dict(encoding=headers.get("encoding",
                                         headers.get("content-encoding")))
        # AWS cannot store empty strings, so we only add these keys if
        # they're present to avoid empty strings.
        for name in ["encryption", "encryption-key", "crypto-key"]:
            if name in headers:
                # NOTE: The client code expects all header keys to be lower
                # case and s/-/_/.
                data[name.lower().replace("-", "_")] = headers[name]
        self.headers = data

    @property
    def sort_key(self):
        # type: () -> str
        """Return an appropriate sort_key for this notification

        For new messages:

            02:{sortkey_timestamp}:{chid}

        For topic messages:

            01:{chid}:{topic}

        Old format for non-topic messages that is no longer returned:

            {chid}:{message_id}

        """
        chid = normalize_id(self.channel_id)
        if self.topic:
            return "01:{chid}:{topic}".format(chid=chid, topic=self.topic)
        elif self.legacy:
            return "{chid}:{message_id}".format(
                chid=chid, message_id=self.message_id
            )
        else:
            # Created as late as possible when storing a message
            self.sortkey_timestamp = self.sortkey_timestamp or ns_time()
            return "02:{sortkey_timestamp}:{chid}".format(
                sortkey_timestamp=self.sortkey_timestamp,
                chid=chid,
            )

    @staticmethod
    def parse_sort_key(sort_key):
        # type: (str) -> Dict[str, Any]
        """Parse the sort key from the database"""
        topic = None
        sortkey_timestamp = None
        message_id = None
        if sort_key.startswith("01:"):
            api_ver, channel_id, topic = sort_key.split(":")
        elif sort_key.startswith("02:"):
            api_ver, raw_sortkey, channel_id = sort_key.split(":")
            sortkey_timestamp = int(raw_sortkey)
        else:
            channel_id, message_id = sort_key.split(":")
            api_ver = "00"
        return dict(api_ver=api_ver, channel_id=channel_id,
                    topic=topic, message_id=message_id,
                    sortkey_timestamp=sortkey_timestamp)

    @property
    def location(self):
        # type: () -> str
        """Return an appropriate value for the Location header"""
        return self.message_id

    @property
    def data_length(self):
        """Return the length of the data"""
        return len(self.data or "")

    def expired(self, at_time=None):
        # type: (Optional[int]) -> bool
        """Indicates whether the message has expired or not

        :param at_time: Optional time to compare for expiration

        """
        now = at_time or int(time.time())
        return now >= ((self.ttl or 0) + self.timestamp)

    @classmethod
    def from_message_table(cls, uaid, item):
        # type: (uuid.UUID, Dict[str, Any]) -> WebPushNotification
        """Create a WebPushNotification from a message table item"""
        key_info = cls.parse_sort_key(item["chidmessageid"])
        if key_info["api_ver"] in ["01", "02"]:
            key_info["message_id"] = item["updateid"]
        notif = cls(
            uaid=uaid,
            channel_id=uuid.UUID(key_info["channel_id"]),
            data=item.get("data"),
            headers=item.get("headers"),
            ttl=item.get("ttl", 0),
            topic=key_info.get("topic"),
            message_id=key_info["message_id"],
            update_id=item.get("updateid"),
            timestamp=item.get("timestamp"),
            sortkey_timestamp=key_info.get("sortkey_timestamp"),
            source="Stored"
        )
        # Ensure we generate the sort-key properly for legacy messges
        if key_info["api_ver"] == "00":
            notif.legacy = True

        return notif

    @classmethod
    def from_webpush_request_schema(cls, data, fernet, legacy=False):
        # type: (Dict[str, Any], Fernet, bool) -> WebPushNotification
        """Create a WebPushNotification from a validated WebPushRequestSchema

        This is a blocking call.
        """
        sub = data["subscription"]
        notif = cls(
            uaid=sub["uaid"],
            channel_id=sub["chid"],
            data=data["body"],
            headers=data["headers"],
            ttl=data["headers"]["ttl"],
            topic=data["headers"]["topic"],
            legacy=legacy,
        )

        if notif.data:
            notif.cleanup_headers()
        else:
            notif.headers = None

        notif.generate_message_id(fernet)
        return notif

    @classmethod
    def from_message_id(cls, message_id, fernet):
        # type: (str, Fernet) -> WebPushNotification
        """Create a WebPushNotification from a message_id

        This is a blocking call.

        The resulting WebPushNotification is not a complete one
        from the database, but has all the parsed attributes
        available that can be derived from the message_id.

        This is suitable for passing to delete calls.

        """
        decrypted_message_id = fernet.decrypt(message_id)
        key_info = cls.parse_decrypted_message_id(decrypted_message_id)
        notif = cls(uaid=uuid.UUID(key_info["uaid"]),
                    channel_id=uuid.UUID(key_info["chid"]),
                    data=None,
                    ttl=None,
                    topic=key_info["topic"],
                    message_id=message_id,
                    sortkey_timestamp=key_info.get("sortkey_timestamp"),
                    )
        if key_info["topic"]:
            notif.update_id = message_id
        return notif

    @classmethod
    def from_serialized(cls, uaid, data):
        # type: (uuid.UUID, Dict[str, str]) -> WebPushNotification
        """Create a WebPushNotification from a deserialized JSON dict"""
        notif = cls(uaid=uaid, channel_id=uuid.UUID(data["channelID"]),
                    data=data.get("data"),
                    headers=data.get("headers"),
                    ttl=data.get("ttl"),
                    topic=data.get("topic"),
                    message_id=str(data["version"]),
                    update_id=str(data["version"]),
                    timestamp=data.get("timestamp"),
                    )
        return notif

    @property
    def version(self):
        # type: () -> str
        """Return a 'version' for use with a websocket client

        In our case we use the message-id as its a unique value for every
        message.

        """
        return self.message_id

    def serialize(self):
        # type: () -> Dict[str, Any]
        """Serialize to a dict for delivery to a connection node"""
        payload = dict(
            channelID=normalize_id(self.channel_id),
            version=self.version,
            ttl=self.ttl,
            topic=self.topic,
            timestamp=self.timestamp,
        )
        if self.data:
            payload["data"] = self.data
            payload["headers"] = self.headers
        return payload

    def websocket_format(self):
        # type: () -> Dict[str, Any]
        """Format a notification for a websocket client"""
        # Firefox currently requires channelIDs to be '-' formatted.
        payload = dict(
            messageType="notification",
            channelID=normalize_id(self.channel_id),
            version=self.version,
        )  # type: Dict[str, Any]
        if self.data:
            payload["data"] = self.data
            payload["headers"] = {
                k.replace("-", "_"): v for k, v in self.headers.items()
            }
        return payload class Client(object):
    def __init__(self, app):
        self.app = app
        self.cookies = DecodingSimpleCookie(self.app)

    def get(self, uri, params=None, version="HTTP/1.0", headers=None,
            body=None, remote_ip=None, protocol=None, host=None,
            files=None, connection=None):
        return self.request(
            "GET", uri, params=params, version=version, headers=headers,
            body=body, remote_ip=remote_ip, protocol=protocol, host=host,
            files=files, connection=connection
        )

    def put(self, uri, params=None, version="HTTP/1.0", headers=None,
            body=None, remote_ip=None, protocol=None, host=None,
            files=None, connection=None):
        return self.request(
            "PUT", uri, params=params, version=version, headers=headers,
            body=body, remote_ip=remote_ip, protocol=protocol, host=host,
            files=files, connection=connection
        )

    def post(self, uri, params=None, version="HTTP/1.0", headers=None,
             body=None, remote_ip=None, protocol=None, host=None,
             files=None, connection=None):
        return self.request(
            "POST", uri, params=params, version=version, headers=headers,
            body=body, remote_ip=remote_ip, protocol=protocol, host=host,
            files=files, connection=connection
        )

    def delete(self, uri, params=None, version="HTTP/1.0", headers=None,
               body=None, remote_ip=None, protocol=None, host=None,
               files=None, connection=None):
        return self.request(
            "DELETE", uri, params=params, version=version, headers=headers,
            body=body, remote_ip=remote_ip, protocol=protocol, host=host,
            files=files, connection=connection
        )

    def head(self, uri, params=None, version="HTTP/1.0", headers=None,
             body=None, remote_ip=None, protocol=None, host=None,
             files=None, connection=None):
        return self.request(
            "HEAD", uri, params=params, version=version, headers=headers,
            body=body, remote_ip=remote_ip, protocol=protocol, host=host,
            files=files, connection=connection
        )

    @inlineCallbacks
    def request(self, method, uri, *args, **kwargs):
        params = kwargs.pop("params", {}) or {}
        if method in ["GET", "HEAD", "OPTIONS"] and params:
            uri = uri + "?" + urllib.urlencode(params)
        elif method in ["POST", "PATCH", "PUT"]\
                and params and not kwargs['body']:
            kwargs['body'] = urllib.urlencode(params)  # pragma: no cover
        connection = kwargs.pop('connection')
        if not connection:
            connection = HTTPConnection()
            connection.xheaders = False
            kwargs['connection'] = connection
        connection.factory = self.app
        cookie_value = self.cookies.output(header="")
        if cookie_value.strip():
            if kwargs['headers'] is None:
                kwargs['headers'] = {}
            kwargs['headers']['Cookie'] = cookie_value.strip()
        request = HTTPRequest(method, uri, *args, **kwargs)
        for k, p in params.items():
            request.arguments.setdefault(k, []).append(p)
        connection.connectionMade()
        connection._request = request
        connection.transport = proto_helpers.StringTransport()
        request.remote_ip = connection.transport.getHost().host
        handler = self.app(request)

        def setup_response():
            headers = HTTPHeaders()
            for line in handler._generate_headers().split("\r\n"):
                if line.startswith("HTTP") or not line.strip():
                    continue
                headers.parse_line(line)
            for cookie in headers.get_list("Set-Cookie"):
                self.cookies.load(cookie)
            response_body = connection.transport.io.getvalue()
            handler.content = response_body.split("\r\n\r\n", 1)[1]
            handler.headers = headers

        if handler._finished:
            setup_response()
            returnValue(handler)
        yield connection.notifyFinish()
        setup_response()
        returnValue(handler) def make_webpush_notification(uaid, chid, ttl=100):
    message_id = str(uuid.uuid4())
    return WebPushNotification(
        uaid=uuid.UUID(uaid),
        channel_id=uuid.UUID(chid),
        update_id=message_id,
        message_id=message_id,
        ttl=ttl,
    ) USER_RECORD_VERSION = 1
DEFAULT_WS_ERR = "http://autopush.readthedocs.io/en/" \
                 "latest/api/websocket.html#private-http-endpoint"
NACK_CODES = range(301, 304)
def extract_code(data):
def log_exception(func):
    def wrapper(self, *args, **kwargs):
    def logging_data(self):
    def from_request(cls, request, **kwargs):
    def __attrs_post_init__(self):
    def user_agent(self):
    def reset_uaid(self):
    def reset_uaid(self, value):
    def uaid_obj(self):
    def uaid_hash(self):
    def uaid(self):
    def uaid(self, value):
    def init_connection(self):
    def pauseProducing(self):
    def resumeProducing(self):
    def stopProducing(self):
    def conf(self):
    def db(self):
    def metrics(self):
    def deferToThread(self, func, *args, **kwargs):
        def f(result):
    def deferToLater(self, when, func, *args, **kwargs):
        def cancel(d):
        def f():
    def trap_cancel(self, fail):
    def trap_connection_err(self, fail):
    def trap_boto3_err(self, fail):
    def force_retry(self, func, *args, **kwargs):
        def wrapper(result, *w_args, **w_kwargs):
    def base_tags(self):
    def log_failure(self, failure, **kwargs):
    def paused(self):
    def _sendAutoPing(self):
    def sendClose(self, code=None, reason=None):
    def nukeConnection(self):
    def onConnect(self, request):
    def processHandshake(self):
    def onMessage(self, payload, isBinary):
    def timeoutConnection(self):
    def onAutoPingTimeout(self):
    def onClose(self, wasClean, code, reason):
    def cleanUp(self, wasClean, code, reason):
    def _save_webpush_notif(self, notif):
    def _lookup_node(self, results):
    def _trap_uaid_not_found(self, fail):
    def _notify_node(self, result):
    def returnError(self, messageType, reason, statusCode, close=True,
                    url=DEFAULT_WS_ERR):
    def error_overload(self, failure, message_type, disconnect=True):
    def error_finish_overload(self, message_type):
    def sendJSON(self, body):
    def process_hello(self, data):
    def _register_user(self, existing_user=True):
    def _verify_user_record(self):
    def error_hello(self, failure):
    def _check_other_nodes(self, result, url=DEFAULT_WS_ERR):
    def finish_hello(self, previous):
    def process_notifications(self):
    def webpush_fetch(self):
    def error_notifications(self, fail):
    def error_notification_overload(self, fail):
    def error_message_overload(self, fail):
    def finish_notifications(self, notifs):
    def finish_webpush_notifications(self, result):
    def _rotate_message_table(self):
    def _monthly_transition(self):
    def _finish_monthly_transition(self, result):
    def error_monthly_rotation_overload(self, fail):
    def _send_ping(self):
    def process_ping(self):
    def process_register(self, data):
    def error_register(self, fail):
    def finish_register(self, endpoint, chid):
    def send_register_finish(self, result, endpoint, chid):
    def process_unregister(self, data):
    def ack_update(self, update):
    def _handle_webpush_ack(self, chid, version, code):
        def ver_filter(notif):
    def _handle_webpush_update_remove(self, result, chid, notif):
    def process_ack(self, data):
    def process_nack(self, data):
    def check_missed_notifications(self, results, resume=False):
    def bad_message(self, typ, message=None, url=DEFAULT_WS_ERR):
    def send_notification(self, update):
    def emit_send_metrics(self, notif):
    def __init__(self, conf, db, agent, clients):
    def put(self, uaid):
    def put(self, uaid, *args):
    def delete(self, uaid, connected_at):
    def __init__(self, conf, ws_factory):
    def ssl_cf(self):
    def __init__(self, resource):
    def getChild(self, path, request):
    def render(self, request):  # pragma: nocover
    def render(self, request):
class SessionStatistics(object):
class PushState(object):
class PushServerProtocol(WebSocketServerProtocol, policies.TimeoutMixin):
class PushServerFactory(WebSocketServerFactory):
class RouterHandler(BaseHandler):
class NotificationHandler(BaseHandler):
class ConnectionWSSite(Site):
class DefaultResource(Resource):
class StatusResource(Resource): def base64url_encode(string):
    # type: (str) -> str
    """Encodes an unpadded Base64 URL-encoded string per RFC 7515."""
    return base64.urlsafe_b64encode(string).strip('=') def ms_time():
    # type: () -> int
    """Return current time.time call as ms and a Python int"""
    return int(time.time() * MS_MULT)
import json
import datetime
import time
import uuid
import twisted.internet.base
import pytest
import autopush.db as db
import autopush.tests
from hashlib import sha256
from collections import defaultdict
from urllib3.exceptions import ConnectTimeoutError
from autobahn.twisted.util import sleep
from autobahn.websocket.protocol import ConnectionRequest
from botocore.exceptions import ClientError
from mock import Mock, patch
from twisted.internet import reactor
from twisted.internet.defer import (
    inlineCallbacks,
    returnValue,
    Deferred
)
from twisted.internet.error import ConnectError
from twisted.trial import unittest
from twisted.web.client import Agent
from autopush.config import AutopushConfig
from autopush.db import DatabaseManager
from autopush.http import InternalRouterHTTPFactory
from autopush.metrics import SinkMetrics
from autopush.utils import WebPushNotification
from autopush.tests.client import Client
from autopush.tests.test_db import make_webpush_notification
from autopush.websocket import (
    PushState,
    PushServerFactory,
    RouterHandler,
    NotificationHandler,
    WebSocketServerProtocol,
)
from autopush.utils import base64url_encode, ms_time
        from twisted.logger import Logger
        def throw(*args, **kwargs):
            raise Exception("Krikey")

        twisted.internet.base.DelayedCall.debug = True
        self._connect()
        msg_mock = Mock(spec=db.Message)
        msg_mock.fetch_messages = Mock(
            side_effect=throw)
        msg_mock.fetch_timestamp_messages = Mock(
            side_effect=throw)
        self.proto.db.message_table = Mock(return_value=msg_mock)
        self.proto.ps.uaid = uuid.uuid4().hex

        self.proto.process_notifications()
        notif_d = self.proto.ps._notification_fetch
        # Tag on our own to follow up
        d = Deferred()

        # Ensure we catch error outs from either call
        notif_d.addErrback(lambda x: d.errback(x))

        def wait(result):
            fail = self.proto.log.failure
            assert fail.called
            assert fail.call_args[1].get('failure').value[0] == 'Krikey'
            d.callback(True)

        self.proto.ps._notification_fetch.addCallback(wait)
        self.proto.ps._notification_fetch.addErrback(lambda x: d.errback(x))
        return d

    def test_process_notifications_provision_err(self):

        def raise_condition(*args, **kwargs):
            raise ClientError(
                {'Error': {'Code': 'ProvisionedThroughputExceededException'}},
                'mock_update_item'
            )

        twisted.internet.base.DelayedCall.debug = True
        self._connect()
        msg_mock = Mock(spec=db.Message)
        msg_mock.fetch_messages = Mock(
            side_effect=raise_condition)
        msg_mock.fetch_timestamp_messages = Mock(
            side_effect=raise_condition)
        self.proto.db.message_table = Mock(return_value=msg_mock)
        self.proto.deferToLater = Mock()
        self.proto.ps.uaid = uuid.uuid4().hex

        self.proto.process_notifications()
        notif_d = self.proto.ps._notification_fetch
        # Tag on our own to follow up
        d = Deferred()

        # Ensure we catch error outs from either call
        notif_d.addErrback(lambda x: d.errback(x))

        def wait(result):
            assert self.proto.deferToLater.called, "Defer not called"
            d.callback(True)

        self.proto.ps._notification_fetch.addCallback(wait)
        self.proto.ps._notification_fetch.addErrback(lambda x: d.errback(x))
        return d

    def test_process_notif_doesnt_run_with_webpush_outstanding(self):
        self._connect()
        self.proto.ps.uaid = dummy_uaid.hex
        self.proto.ps.updates_sent[dummy_chid_str] = [dummy_notif()]
        self.proto.deferToLater = Mock()
        self.proto.process_notifications()
        assert self.proto.deferToLater.called
        assert self.proto.ps._notification_fetch is None

    def test_process_notif_doesnt_run_when_paused(self):
        self._connect()
        self.proto.ps.uaid = uuid.uuid4().hex
        self.proto.ps.pauseProducing()
        with patch("autopush.websocket.reactor") as mr:
            self.proto.process_notifications()
            assert mr.callLater.mock_calls > 0

    def test_process_notif_doesnt_run_after_stop(self):
        self._connect()
        self.proto.ps.uaid = uuid.uuid4().hex
        self.proto.ps._should_stop = True
        self.proto.process_notifications()
        assert self.proto.ps._notification_fetch is None

    def test_check_notif_doesnt_run_after_stop(self):
        self._connect()
        self.proto.ps.uaid = uuid.uuid4().hex
        self.proto.ps._should_stop = True
        self.proto.check_missed_notifications(None)
        assert self.proto.ps._notification_fetch is None

    def test_process_notif_paused_on_finish(self):
        self._connect()
        self.proto.ps.uaid = uuid.uuid4().hex
        self.proto.ps.pauseProducing()
        with patch("autopush.websocket.reactor") as mr:
            self.proto.finish_notifications(None)
            assert mr.callLater.mock_calls > 0

    def test_notif_finished_with_webpush(self):
        self._connect()
        self.proto.ps.uaid = uuid.uuid4().hex
        self.proto.deferToLater = Mock()
        self.proto.ps._check_notifications = True
        self.proto.ps.scan_timestamps = True
        self.proto.finish_notifications((None, []))
        assert self.proto.deferToLater.called

    def test_notif_finished_with_webpush_with_notifications(self):
        self._connect()
        self.proto.ps.uaid = uuid.uuid4().hex
        self.proto.ps._check_notifications = True
        self.proto.process_notifications = Mock()


        notif = make_webpush_notification(