<commit_before>"""Token based pyramid authentication policy."""
import hashlib
from datetime import datetime

from colander import Invalid
from persistent.dict import PersistentDict
from pyramid.authentication import CallbackAuthenticationPolicy
from pyramid.interfaces import IAuthenticationPolicy
from pyramid.request import Request
from pyramid.traversal import resource_path
from zope.interface import implementer
from zope.interface import Interface
from zope.component import ComponentLookupError

from adhocracy_core.interfaces import ITokenManger
from adhocracy_core.interfaces import IRolesUserLocator
from adhocracy_core.schema import Resource


@implementer(ITokenManger)
class TokenMangerAnnotationStorage:

    """Manage authentication tokens and use object annotation to store them.

    Constructor arguments:

    :param context: the object to annotate the authentication token storage.
    """

    annotation_key = '_tokenmanager_storage'

    def __init__(self, context):
        self.context = context

    @property
    def token_to_user_id_timestamp(self):
        if not hasattr(self.context, self.annotation_key):
            setattr(self.context, self.annotation_key, PersistentDict())
        return getattr(self.context, self.annotation_key)

    def create_token(self, userid: str, secret='', hashalg='sha512') -> str:
        """Create authentication token for user_id.

        :param secret:  the secret used to salt the generated token.
        :param hashalg: Any hash algorithm supported by :func: `hashlib.new`.
                        This is used to create the authentication token.
        """
        timestamp = datetime.now()
        value = self._build_token_value(userid, timestamp, secret)
        token = hashlib.new(hashalg, value).hexdigest()
        self.token_to_user_id_timestamp[token] = (userid, timestamp)
        return token

    def _build_token_value(self, user_id: str, timestamp: datetime,
                           secret: '') -> str:
        time_bytes = timestamp.isoformat().encode('UTF-8')
        secret_bytes = secret.encode('UTF-8', 'replace')
        user_bytes = user_id.encode('UTF-8', 'replace')
        return time_bytes + secret_bytes + user_bytes

    def get_user_id(self, token: str, timeout: float=None) -> str:
        """Get user_id for authentication token.

        :param timeout:  Maximum number of seconds which a newly create token
                        will be considered valid.
                        The `None` value is allowed to disable the timeout.
        :returns: user id for this token
        :raises KeyError: if there is no corresponding user_id
        """
        userid, timestamp = self.token_to_user_id_timestamp[token]
        if self._is_expired(timestamp, timeout):
            del self.token_to_user_id_timestamp[token]
            raise KeyError
        return userid

    def _is_expired(self, timestamp: datetime, timeout: float=None) -> bool:
        if timeout is None:
            return False
        now = datetime.now()
        delta = now - timestamp
        return delta.total_seconds() >= timeout

    def delete_token(self, token: str):
        """Delete authentication token."""
        if token in self.token_to_user_id_timestamp:
            del self.token_to_user_id_timestamp[token]


def get_tokenmanager(request: Request, **kwargs) -> ITokenManger:
    """Adapter request.root to ITokenmanager and return it.

    :returns: :class:'adhocracy_core.interfaces.ITokenManager or None.
    """
    try:
        return ITokenManger(request.root)
    except (ComponentLookupError, TypeError):
        return None


def _get_raw_x_user_headers(request: Request) -> tuple:
    """Return not validated tuple with the X-User-Path/Token values."""
    user_url = request.headers.get('X-User-Path', '')
    # FIXME find a proper solution, user_url/path as userid does not work
    # well with the pyramid authentication system. We don't have the
    # a context or root object to resolve the resource path when processing
    # the unauthenticated_userid and effective_principals methods.
    app_url_length = len(request.application_url)
    user_path = None
    if len(user_url) >= app_url_length:
        user_path = user_url[app_url_length:][:-1]
    if user_url.startswith('/'):
        user_path = user_url
    token = request.headers.get('X-User-Token', None)
    return user_path, token


def _get_x_user_headers(request: Request) -> tuple:
    """Return tuple with the X-User-Path/Token values or (None, None)."""
    schema = Resource().bind(request=request, context=request.context)
    user_url = request.headers.get('X-User-Path', None)
    user_path = None
    try:
        user = schema.deserialize(user_url)
        user_path = resource_path(user)
    except Invalid:
        # FIXME if we want to use multiple authentication policies we should
        # ignore exceptions at all.
        # Else we should raise a proper colander error.
        pass
    token = request.headers.get('X-User-Token', None)
    return (user_path, token)


@implementer(IAuthenticationPolicy)
class TokenHeaderAuthenticationPolicy(CallbackAuthenticationPolicy):

    """A :term:`authentication policy` based on the the X-User-* header.

    To authenticate the client has to send http header with `X-User-Token`
    and `X-User-Path`.
<commit_msg>Make URL/path validation safer.<commit_after>"""Token based pyramid authentication policy."""
import hashlib
from datetime import datetime

from colander import Invalid
from persistent.dict import PersistentDict
from pyramid.authentication import CallbackAuthenticationPolicy
from pyramid.interfaces import IAuthenticationPolicy
from pyramid.request import Request
from pyramid.traversal import resource_path
from zope.interface import implementer
from zope.interface import Interface
from zope.component import ComponentLookupError

from adhocracy_core.interfaces import ITokenManger
from adhocracy_core.interfaces import IRolesUserLocator
from adhocracy_core.schema import Resource


@implementer(ITokenManger)
class TokenMangerAnnotationStorage:

    """Manage authentication tokens and use object annotation to store them.

    Constructor arguments:

    :param context: the object to annotate the authentication token storage.
    """

    annotation_key = '_tokenmanager_storage'

    def __init__(self, context):
        self.context = context

    @property
    def token_to_user_id_timestamp(self):
        if not hasattr(self.context, self.annotation_key):
            setattr(self.context, self.annotation_key, PersistentDict())
        return getattr(self.context, self.annotation_key)

    def create_token(self, userid: str, secret='', hashalg='sha512') -> str:
        """Create authentication token for user_id.

        :param secret:  the secret used to salt the generated token.
        :param hashalg: Any hash algorithm supported by :func: `hashlib.new`.
                        This is used to create the authentication token.
        """
        timestamp = datetime.now()
        value = self._build_token_value(userid, timestamp, secret)
        token = hashlib.new(hashalg, value).hexdigest()
        self.token_to_user_id_timestamp[token] = (userid, timestamp)
        return token

    def _build_token_value(self, user_id: str, timestamp: datetime,
                           secret: '') -> str:
        time_bytes = timestamp.isoformat().encode('UTF-8')
        secret_bytes = secret.encode('UTF-8', 'replace')
        user_bytes = user_id.encode('UTF-8', 'replace')
        return time_bytes + secret_bytes + user_bytes

    def get_user_id(self, token: str, timeout: float=None) -> str:
        """Get user_id for authentication token.

        :param timeout:  Maximum number of seconds which a newly create token
                        will be considered valid.
                        The `None` value is allowed to disable the timeout.
        :returns: user id for this token
        :raises KeyError: if there is no corresponding user_id
        """
        userid, timestamp = self.token_to_user_id_timestamp[token]
        if self._is_expired(timestamp, timeout):
            del self.token_to_user_id_timestamp[token]
            raise KeyError
        return userid

    def _is_expired(self, timestamp: datetime, timeout: float=None) -> bool:
        if timeout is None:
            return False
        now = datetime.now()
        delta = now - timestamp
        return delta.total_seconds() >= timeout

    def delete_token(self, token: str):
        """Delete authentication token."""
        if token in self.token_to_user_id_timestamp:
            del self.token_to_user_id_timestamp[token]


def get_tokenmanager(request: Request, **kwargs) -> ITokenManger:
    """Adapter request.root to ITokenmanager and return it.

    :returns: :class:'adhocracy_core.interfaces.ITokenManager or None.
    """
    try:
        return ITokenManger(request.root)
    except (ComponentLookupError, TypeError):
        return None


def _get_raw_x_user_headers(request: Request) -> tuple:
    """Return not validated tuple with the X-User-Path/Token values."""
    user_url = request.headers.get('X-User-Path', '')
    # FIXME find a proper solution, user_url/path as userid does not work
    # well with the pyramid authentication system. We don't have the
    # a context or root object to resolve the resource path when processing
    # the unauthenticated_userid and effective_principals methods.
    app_url_length = len(request.application_url)
    user_path = None
    if user_url.startswith('/'):
        user_path = user_url
    elif len(user_url) >= app_url_length:
        user_path = user_url[app_url_length:][:-1]
    token = request.headers.get('X-User-Token', None)
    return user_path, token


def _get_x_user_headers(request: Request) -> tuple:
    """Return tuple with the X-User-Path/Token values or (None, None)."""
    schema = Resource().bind(request=request, context=request.context)
    user_url = request.headers.get('X-User-Path', None)
    user_path = None
    try:
        user = schema.deserialize(user_url)
        user_path = resource_path(user)
    except Invalid:
        # FIXME if we want to use multiple authentication policies we should
        # ignore exceptions at all.
        # Else we should raise a proper colander error.
        pass
    token = request.headers.get('X-User-Token', None)
    return (user_path, token)


@implementer(IAuthenticationPolicy)
class TokenHeaderAuthenticationPolicy(CallbackAuthenticationPolicy):

    """A :term:`authentication policy` based on the the X-User-* header.

    To authenticate the client has to send http header with `X-User-Token`
    and `X-User-Path`.

{'fraction_non_alphanumeric': 0.08417132216014897, 'fraction_numerical': 0.0013035381750465549, 'mean_word_length': 3.1858924395947, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2140993', 'n_tokens_mistral': 2970, 'n_tokens_neox': 2867, 'n_words': 905}