1 files changed, 0 insertions, 471 deletions

diff --git a/otp2289/server.py b/otp2289/server.py
deleted file mode 100644
index 7f742cd..0000000
--- a/otp2289/server.py
+++ /dev/null
@@ -1,471 +0,0 @@
-# -*- coding: utf-8 -*-
-# SPDX-License-Identifier: BSD-2-Clause-FreeBSD
-#
-# Copyright (c) 2020-2022 Simeon Simeonov
-# All rights reserved.
-#
-# Redistribution and use in source and binary forms, with or without
-# modification, are permitted provided that the following conditions
-# are met:
-# 1. Redistributions of source code must retain the above copyright
-#    notice, this list of conditions and the following disclaimer.
-# 2. Redistributions in binary form must reproduce the above copyright notice,
-#    this list of conditions and the following disclaimer in the documentation
-#    and/or other materials provided with the distribution.
-#
-# THIS SOFTWARE IS PROVIDED BY THE AUTHORS ``AS IS'' AND ANY EXPRESS OR
-# IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
-# OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
-# IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
-# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
-# NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
-# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
-# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
-# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
-# THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-"""A pure Python implementation of the RFC-2289 OTP server"""
-import binascii
-import hashlib
-
-from .generator import OTP_ALGO_MD5, OTPGenerator, OTPGeneratorException
-
-
-class OTPStateException(Exception):
-    """OTPStateException class"""
-
-
-class OTPStoreException(Exception):
-    """OTPStoreException class"""
-
-
-class OTPInvalidResponse(Exception):
-    """OTPInvalidResponse class"""
-
-
-class OTPState:
-    """
-    OTPState class
-
-    The OTPState class represents a single state on the server side that can:
-    - generate a challenge
-    - validate the corresponding generated response from the generator
-    """
-
-    def __init__(
-        self,
-        ot_hex: str,
-        current_step: int,
-        seed: str,
-        hash_algo=OTP_ALGO_MD5,
-    ):
-        """
-        Constructs an OTPState object with the given arguments.
-
-        Keyword Arguments:
-        :param ot_hex: The one-time hex from the last successful authentication
-                       or None for a newly initialized sequence.
-        :type ot_hex: str or None
-
-        :param current_step: The current step that is sent with the challenge
-        :type current_step: int
-
-        :param seed: The seed that is sent with the challenge
-        :type seed: str
-
-        :param hash_algo: The hash algo, defaults to OTP_ALGO_MD5
-        :type hash_algo: int or str
-
-        :raises otp2289.OTPStateException: If the input does not validate
-        """
-        # enforce the rfc2289 constraints
-        try:
-            self._seed = OTPGenerator.validate_seed(seed)
-            self._hash_algo = OTPGenerator.validate_hash_algo(hash_algo)
-            self._step = OTPGenerator.validate_step(current_step)
-        except OTPGeneratorException as exp:
-            raise OTPStateException(exp.args[0]) from None
-        self._current_digest = None
-        if ot_hex is not None:
-            self._current_digest = self.validate_hex(ot_hex)
-        self._new_digest_hex = None  # set upon a successful validation
-
-    def __repr__(self):
-        """repr implementation"""
-        return (
-            f'{self.__class__} at {id(self)} '
-            f'(ot_hex={self._current_digest}, current_step={self._step}, '
-            f'seed={self._seed}, '
-            f'hash_algo={self._hash_algo})'
-        )
-
-    @property
-    def challenge_string(self) -> str:
-        """challenge_string-property"""
-        # RFC-2289: "...the entire challenge string MUST be
-        # terminated with either a space or a new line."
-        return f'otp-{self._hash_algo} {self._step} {self._seed} '
-
-    @property
-    def current_digest(self) -> bytes:
-        """current_digest-property"""
-        return self._current_digest
-
-    @property
-    def hash_algo(self) -> str:
-        """hash_algo-property"""
-        return self._hash_algo
-
-    @property
-    def seed(self) -> str:
-        """seed-property"""
-        return self._seed
-
-    @property
-    def step(self) -> int:
-        """step-property"""
-        return self._step
-
-    @property
-    def validated(self) -> bool:
-        """validated-property"""
-        return bool(self._new_digest_hex)
-
-    @classmethod
-    def from_dict(cls, dict_obj: dict):
-        """
-        Returns an OTPState object from the dict-object
-
-        :param dict_obj: The dict object
-        :type dict_obj: dict
-
-        :return: A new OTPState object
-        :rtype: otp2289.OTPStore
-        """
-        return cls(**dict_obj)
-
-    @staticmethod
-    def response_to_bytes(response: str) -> bytes:
-        """
-        A wrapper that handles/validates the response as specified by RFC-2289.
-
-        The method first checks if response is a token and tries to convert
-        it to bytes. If that fails, the method assumes that response is a hex.
-        If neither of those attempts succeeds OTPInvalidResponse is raised.
-        It is up to the caller to run another iteration and compare the result
-        to an existing digest in this state.
-
-        :param response: The response to this state (its challenge)
-        :type response: str
-
-        :raises otp2289.OTPInvalidResponse: If the response is corrupt/illegal,
-                                            but not if it simply does not
-                                            validate
-
-        :return: The bytes representation of response (if any)
-        :rtype: bytes
-        """
-        try:
-            return OTPGenerator.tokens_to_bytes(response)
-        except OTPGeneratorException:
-            # now assume hex...
-            try:
-                return OTPState.validate_hex(response)
-            except OTPStateException:
-                raise OTPInvalidResponse(
-                    'The response is neither a valid token or hex'
-                ) from None
-
-    @staticmethod
-    def validate_hex(ot_hex: str) -> bytes:
-        """
-        Validates the provided hexidigest.
-
-        :param ot_hex: The one-time hex to validate
-        :type ot_hex: str
-
-        :raises otp2289.OTPStateException: If hex does not validate
-
-        :return: The validated hex (without leading 0x) converted to bytes
-        :rtype: bytes
-        """
-        if not isinstance(ot_hex, str):
-            raise OTPStateException('OT-hex must be a str')
-        if ot_hex.startswith('0x'):
-            ot_hex = ot_hex[2:]
-            ot_hex = ot_hex.strip().lower()
-        if len(ot_hex) != 16:
-            raise OTPStateException(
-                'The length of the hex should be 16 '
-                '(representing 64 bits digest)'
-            )
-        try:
-            return binascii.unhexlify(ot_hex)
-        except binascii.Error:
-            raise OTPStateException('Invalid OT-hex') from None
-
-    def get_next_state(self):
-        """
-        Returns the next state for a validated OTPState.
-
-        This is a brand new OTPState object with the same hash_algo and seed
-        where step -= 1 and ot_hex = self._new_digest_hex
-
-        :return: The next OTPState if validated, None otherwise
-        :rtype: otp2289.OTPState or None
-        """
-        if self._new_digest_hex is None:
-            return None
-        return OTPState(
-            self._new_digest_hex,
-            self._step - 1,
-            self._seed,
-            self._hash_algo,
-        )
-
-    def response_validates(
-        self,
-        response: str,
-        store_valid_response: str = True,
-    ) -> bool:
-        """
-        Validates the incoming response as specified by RFC-2289.
-
-        :param response: The response to this state (its challenge)
-        :type response: str
-
-        :param store_valid_response: Should a valid response be stored
-        :type store_valid_response: bool
-
-        :raises otp2289.OTPInvalidResponse: If the response does not match
-                                            this state
-
-        :return: Returns True if response validates, False otherwise
-        :rtype: bool
-        """
-        # self.response_to_bytes raises OTPInvalidResponse in case response
-        # is corrupt or in a wrong format
-        response_bytes = self.response_to_bytes(response)
-        if self._hash_algo == 'md5':
-            digest = hashlib.md5(response_bytes).digest()
-            if (
-                self._current_digest is None
-                or OTPGenerator.strxor(digest[0:8], digest[8:])
-                == self._current_digest
-            ):
-                if store_valid_response:
-                    self._new_digest_hex = binascii.hexlify(
-                        response_bytes
-                    ).decode()
-                return True
-            return False
-        if self._hash_algo == 'sha1':
-            digest = hashlib.sha1(response_bytes).digest()
-            if (
-                self._current_digest is None
-                or OTPGenerator.sha1_digest_folding(
-                    hashlib.sha1(response_bytes).digest()
-                )
-                == self._current_digest
-            ):
-                if store_valid_response:
-                    self._new_digest_hex = binascii.hexlify(
-                        response_bytes
-                    ).decode()
-                return True
-            return False
-        # this should not happen since the hash_algo is validated by the caller
-        raise OTPInvalidResponse(f'Ivalid hash_algo: {self._hash_algo}')
-
-    def to_dict(self) -> dict:
-        """
-        Returns a dict representation of the object.
-
-        This could be the base for a JSON serialization.
-
-        :return: The dict representation of the object
-        :rtype: dict
-        """
-        ot_hex = self._current_digest
-        if ot_hex is not None:
-            ot_hex = binascii.hexlify(self._current_digest).decode()
-        return {
-            'ot_hex': ot_hex,
-            'current_step': self._step,
-            'seed': self._seed,
-            'hash_algo': self._hash_algo,
-        }
-
-
-class OTPStore:
-    """
-    OTPStore class
-
-    A helper / container class that stores OTPState objects in a 2 layered
-    dict structure represented by [domain][key].
-
-    The class could serve as a base class when implementing store backends.
-    """
-
-    def __init__(self, data=None):
-        """
-        Constructs an OTPStore object from data
-
-        :param data: The data object, defaults to None
-        :type data: object or None
-        """
-        self._data = {}  # {key1: {state1-data...}, key2: {state2-data...}}
-        self._states = {}  # OTPState: (domain, key) - dict
-        if data is not None:
-            self._add_data(data)
-
-    def __contains__(self, state):
-        """membership test"""
-        return state in self._states
-
-    def __iter__(self):
-        """iterator for OTPStore"""
-        return iter(self._data)
-
-    def __len__(self):
-        """len() implementation"""
-        return len(self._data)
-
-    @property
-    def data(self) -> dict:
-        """
-        data-property
-
-        Exposes the entire raw-data structure (dict).
-        Use the high level methods when possible!
-        """
-        return self._data
-
-    @property
-    def states(self) -> dict:
-        """
-        states-property
-
-        Exposes the entire states structure (dict).
-        Use the high level methods when possible!
-        """
-        return self._states
-
-    def add_state(self, key: str, state: OTPState):
-        """
-        Adds an OTPState object with a given key.
-
-        :param key: The key under which to add the state
-        :type key: str
-
-        :param state: The OTPState object
-        :type state: otp2289.OTPState
-
-        :raises otp2289.OTPStoreException: On failure
-        """
-        if not isinstance(key, str):
-            raise OTPStoreException('key must be a str')
-        if not isinstance(state, OTPState):
-            raise OTPStoreException('state must be an OTPState-object')
-        self._data[key] = state
-        self._states[state] = key
-
-    def get(self, key, default=None):
-        """A wrapper for dict.get"""
-        return self._data.get(key, default)
-
-    def items(self):
-        """A wrapper for dict.items"""
-        return self._data.items()
-
-    def pop_state(self, key: str) -> OTPState:
-        """
-        Removes specified key and returns the corresponding OTPState-object.
-
-        :param key: The key
-        :type key: str
-
-        :raises KeyError: If key does not exist
-
-        :raises otp2289.OTPStoreException: On failure
-
-        :return: The state corresponding to the key
-        :rtype: otp2289.OTPState
-        """
-        if not isinstance(key, str):
-            raise OTPStoreException('key must be a str')
-        state = self._data.pop(key)
-        self._states.pop(state)
-        return state
-
-    def response_validates(
-        self,
-        key: str,
-        response: str,
-        store_valid_response: bool = True,
-    ) -> bool:
-        """
-        A method that wraps around OTPState.response_validates and
-        OTPState.get_next_state.
-
-        The response is validated against the OTPState object that corresponds
-        to key (if any). If store_valid_response is True, the state is replaced
-        by the next state on successful validation.
-
-        :param key: The key
-        :type key: str
-
-        :param response: The response to this state (its challenge)
-        :type response: str
-
-        :param store_valid_response: Should a valid response be stored
-        :type store_valid_response: bool
-
-        :raises KeyError: If the key is not present
-
-        :raises otp2289.OTPInvalidResponse: If the response does not match
-                                            this state
-
-        :return: Returns True if response validates, False otherwise
-        :rtype: bool
-        """
-        state = self._data[key]
-        rvalue = state.response_validates(response, store_valid_response)
-        if rvalue and store_valid_response:
-            next_state = state.get_next_state()
-            self._data[key] = next_state
-            self._states[next_state] = key
-            self._states.pop(state)
-        return rvalue
-
-    def to_dict(self) -> dict:
-        """
-        Returns a dict representation of the object.
-
-        This could be the base for a JSON serialization.
-
-        :return: The dict representation of the object
-        :rtype: dict
-        """
-        return {key: state.to_dict() for key, state in self._data.items()}
-
-    def _add_data(self, dict_obj: dict) -> dict:
-        """
-        Adds data from a dict object (dict_obj).
-
-        This method should probably be either overloaded or wrapped
-        in a child class.
-
-        dict_obj has the following format:
-        {'key': {'ot_hex': val1,
-                 'current_step': val2,
-                 'seed': val3,
-                 'hash_algo': val4},
-         ...., ....}
-
-        :param dict_obj: The dict-object
-        :type dict_obj: dict
-        """
-        if not dict_obj:
-            return
-        for key, state_dict in dict_obj.items():
-            self.add_state(key, OTPState(**state_dict))