akamai.py

# -*- coding: utf-8 -*-

import time

from Crypto.Hash import HMAC, SHA256


class AkamaiSecureToken(object):
    """
    Gera um assinatura para ser utilizada junto com o Akamai, para autenticar
    o usuário e liberar acesso ao stream.

    >>> AkamaiSecureToken(start_time=123, key='abcd').encrypted_token
    'st=123~exp=423~acl=%2f*~hmac=c9d583d364a7d83aa75e0896f7a864aca2e470b009caf700d601da365d44c831'

    >>> AkamaiSecureToken(start_time=123, ip='127.0.0.1', key='abcd').encrypted_token
    'ip=127.0.0.1~st=123~exp=423~acl=%2f*~hmac=7d807cbc347ae77142215d27859d3ef95d1998230f05a45d953b14aba6987e58'

    """
    attributes = {
        'start_time': int(time.time()),
        'window': 300,
        'algo': 'sha256',
        'key': 'aabbccddeeff00112233445566778899',
        'field_delimiter': '~',
        'early_url_encoding': False,
        'debug': False,
        'acl': None,
        'url': None,
        'ip': None,
        'session_id': None,
        'data': None,
        'salt': None
    }

    fields = {
        'start_time': 'st',
        'expiration_field': 'exp',
        'acl': 'acl',
        'url': 'url',
        'ip': 'ip',
        'session_id': 'id',
        'data': 'data',
        'salt': 'salt',
    }

    def __init__(self, *args, **kwargs):
        for field, value in kwargs.items():
            if field not in self.attributes:
                raise ValueError('{} is not a valid field'.format(field))

            self.attributes[field] = value

    def __getattr__(self, field_name):
        if field_name.endswith('_field'):
            field = field_name.replace('_field', '')
            field_synonym = self.fields[field]
            value = self.attributes[field]

            if value:
                return '{field_synonym}={value}{field_delimiter}'.format(
                    field_synonym=field_synonym,
                    value=value,
                    field_delimiter=self.field_delimiter
                )

            return ''

        return self.attributes[field_name]

    @property
    def expiration_field(self):
        return 'exp={expiration_time}{field_delimiter}'.format(
            expiration_time=self.start_time + self.window,
            field_delimiter=self.field_delimiter
        )

    @property
    def acl_field(self):
        if not self.acl and not self.url:
            return '{field_acl}=/*{field_delimiter}'.format(
                field_acl=self.fields['acl'],
                field_delimiter=self.field_delimiter
            )

        return '{field_acl}={acl}{field_delimiter}'.format(
            field_acl=self.fields['acl'],
            acl=self.acl,
            field_delimiter=self.field_delimiter
        )

    @property
    def raw_token(self):
        token = '{ip_field}{start_time_field}{expiration_field}' \
                '{acl_field}{session_id_field}{data_field}'

        token = token.format(
            ip_field=self.ip_field,
            start_time_field=self.start_time_field,
            expiration_field=self.expiration_field,
            acl_field=self.acl_field,
            session_id_field=self.session_id_field,
            data_field=self.data_field,
        )

        return token

    @property
    def token_digest(self):
        token = '{raw_token}{url_field}{salt_field}'

        token = token.format(
            raw_token=self.raw_token,
            url_field=self.url_field,
            salt_field=self.salt_field,
        )
        token = token.rstrip(self.field_delimiter)

        return token

    @property
    def key(self):
        return self.attributes['key'].replace(' ', '').decode('hex')

    @property
    def encrypted_token(self):
        hmac = HMAC.new(key=self.key, msg=self.token_digest, digestmod=SHA256).hexdigest()

        token = '{raw_token}hmac={hmac}'.format(
            raw_token=self.raw_token,
            hmac=hmac
        )

        token = token.replace('/', '%2f')

        return token

if __name__ == "__main__":
    import doctest
    doctest.testmod()