From 12c5e38691b24d12a8fc9605e3b93fac86d461ae Mon Sep 17 00:00:00 2001 From: Simeon Simeonov Date: Mon, 21 Mar 2022 13:25:26 +0100 Subject: Reformat the code with annotations and new code style --- README.md | 28 ++++------ otp2289/__init__.py | 51 +++++++++-------- otp2289/__main__.py | 155 ++++++++++++++++++++++++++++++++++----------------- otp2289/generator.py | 142 ++++++++++++++++++++++++++-------------------- otp2289/server.py | 122 ++++++++++++++++++++++++---------------- 5 files changed, 297 insertions(+), 201 deletions(-) diff --git a/README.md b/README.md index 7ba6487..a3b00c5 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # pyotp2289 -pyotp2289 is a pure Python 3 implementation of "A One-Time Password System" - +*pyotp2289* is a pure Python 3 implementation of "A One-Time Password System" - RFC-2289. It requires no additional libraries. @@ -30,7 +30,7 @@ I hope that somebody will find it useful. ### FreeBSD -`pyotp2289` is included in the official ports-tree. +*pyotp2289* is included in the official ports-tree. ```bash cd /usr/ports/security/py-pyotp2289 @@ -41,21 +41,13 @@ I hope that somebody will find it useful. ### Gentoo ```bash - layman -a sgs - emerge dev-python/pyotp2289 - ``` - - -### Fedora ( >= 31) + # add sgs' custom repository using app-eselect/eselect-repository + eselect repository add sgs -Set up a custom repo as described: https://pkg.pichove.org/Fedora/README.txt - -Key fingerprint: A664 5797 661E 2F47 3DD3 FF06 BCE7 0555 C3BB 08F7 - -Install the package: + # ... or using app-portage/layman (obsolete) + layman -a sgs - ```bash - sudo dnf install python3-pyotp2289 + emerge dev-python/pyotp2289 ``` @@ -189,7 +181,7 @@ a complete reference. If you don't care about developing applications in Python and only care about generating one-time passwords (tokens / hex digests) and authenticating with -existing solutions (f.i. FreeBSD servers), pyotp2289 comes with a simple CLI: +existing solutions (f.i. FreeBSD servers), *pyotp2289* comes with a simple CLI: ```bash python -m otp2289 --generate-otp-response -f token -i 498 -s TesT @@ -207,7 +199,7 @@ starting from (and including) 498. ## Support and contributing -pyotp2289 is hosted on GitHub: https://github.com/blackm0re/pyotp2289 +*pyotp2289* is hosted on GitHub: https://github.com/blackm0re/pyotp2289 ## Author @@ -217,7 +209,7 @@ Simeon Simeonov - sgs @ LiberaChat ## [License](https://github.com/blackm0re/pyotp2289/blob/master/LICENSE) -Copyright (c) 2020, Simeon Simeonov +Copyright (c) 2020-2022 Simeon Simeonov All rights reserved. [Licensed](https://github.com/blackm0re/pyotp2289/blob/master/LICENSE) under the BSD 2-clause. diff --git a/otp2289/__init__.py b/otp2289/__init__.py index 36d1659..608595b 100644 --- a/otp2289/__init__.py +++ b/otp2289/__init__.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- # SPDX-License-Identifier: BSD-2-Clause-FreeBSD # -# Copyright (c) 2020, Simeon Simeonov +# Copyright (c) 2020-2022 Simeon Simeonov # All rights reserved. # # Redistribution and use in source and binary forms, with or without @@ -24,20 +24,23 @@ # (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 RFC-2289""" -from .generator import (OTP_ALGO_MD5, - OTP_ALGO_SHA1, - OTPChallengeException, - OTPGenerator, - OTPGeneratorException) -from .server import (OTPInvalidResponse, - OTPState, - OTPStateException, - OTPStore, - OTPStoreException) - +from .generator import ( + OTP_ALGO_MD5, + OTP_ALGO_SHA1, + OTPChallengeException, + OTPGenerator, + OTPGeneratorException, +) +from .server import ( + OTPInvalidResponse, + OTPState, + OTPStateException, + OTPStore, + OTPStoreException, +) __author__ = 'Simeon Simeonov' -__version__ = '1.0.0' +__version__ = '1.1.0-beta1' __license__ = 'BSD 2-Clause' @@ -51,13 +54,15 @@ def int_or_str(value): VERSION = tuple(map(int_or_str, __version__.split('.'))) -__all__ = ['OTP_ALGO_MD5', - 'OTP_ALGO_SHA1', - 'OTPChallengeException', - 'OTPGenerator', - 'OTPGeneratorException', - 'OTPInvalidResponse', - 'OTPState', - 'OTPStateException', - 'OTPStore', - 'OTPStoreException'] +__all__ = [ + 'OTP_ALGO_MD5', + 'OTP_ALGO_SHA1', + 'OTPChallengeException', + 'OTPGenerator', + 'OTPGeneratorException', + 'OTPInvalidResponse', + 'OTPState', + 'OTPStateException', + 'OTPStore', + 'OTPStoreException', +] diff --git a/otp2289/__main__.py b/otp2289/__main__.py index 7bb0673..471ae24 100644 --- a/otp2289/__main__.py +++ b/otp2289/__main__.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- # SPDX-License-Identifier: BSD-2-Clause-FreeBSD # -# Copyright (c) 2020, Simeon Simeonov +# Copyright (c) 2020-2022 Simeon Simeonov # All rights reserved. # # Redistribution and use in source and binary forms, with or without @@ -35,6 +35,7 @@ python -m otp2289 --generate-otp-response -s TesT -i 499 -f token import argparse import errno import getpass +import io import os import secrets import string @@ -48,7 +49,7 @@ def eprint(*arg, **kwargs): print(*arg, file=sys.stderr, flush=True, **kwargs) -def generate_otp_response(args): +def generate_otp_response(args: argparse.Namespace) -> str: """ Generates a response based on the parameters sent from the parser @@ -65,24 +66,29 @@ def generate_otp_response(args): generator = otp2289.generator.OTPGenerator( args.password.encode(), args.seed, - args.hash_algo) + args.hash_algo, + ) if args.challenge_string: if args.output_format == 'token': return generator.generate_otp_words_from_challenge( - args.challenge_string) + args.challenge_string + ) return generator.generate_otp_hexdigest_from_challenge( - args.challenge_string) + args.challenge_string + ) # regular parameters header = '' if not args.quiet: - header = (f'Seed: {args.seed}, Step: {args.step}, ' - f'Hash: {args.hash_algo}{os.linesep}') + header = ( + f'Seed: {args.seed}, Step: {args.step}, ' + f'Hash: {args.hash_algo}{os.linesep}' + ) if args.output_format == 'token': return header + generator.generate_otp_words(args.step) return header + generator.generate_otp_hexdigest(args.step) -def generate_otp_range(args): +def generate_otp_range(args: argparse.Namespace) -> str: """ Generates range of responses based on the parameters sent from the parser @@ -99,7 +105,8 @@ def generate_otp_range(args): generator = otp2289.generator.OTPGenerator( args.password.encode(), args.seed, - args.hash_algo) + args.hash_algo, + ) if args.output_format == 'token': method = generator.generate_otp_words else: @@ -112,25 +119,35 @@ def generate_otp_range(args): # any need for quiet? header = '' if not args.quiet: - header = (f'Seed: {args.seed}, Step: {args.step}, ' - f'Hash: {args.hash_algo}, Range: {args.range}{os.linesep}') + header = ( + f'Seed: {args.seed}, Step: {args.step}, ' + f'Hash: {args.hash_algo}, Range: {args.range}{os.linesep}' + ) return header + os.linesep.join( - [f'{step}: ' + method(step) for step in range( - args.step, args.step - args.range, -1)]) + [ + f'{step}: ' + method(step) + for step in range( + args.step, + args.step - args.range, + -1, + ) + ] + ) -def get_rnd_seed(): +def get_rnd_seed() -> str: """ Returns a random seed in the format: 2 random letters (capitalize()) + 5 random digits """ rnd = secrets.SystemRandom() - return (''.join(rnd.choices(string.ascii_lowercase, k=2)).capitalize() + - ''.join(rnd.choices(string.digits, k=5))) + return ''.join( + rnd.choices(string.ascii_lowercase, k=2) + ).capitalize() + ''.join(rnd.choices(string.digits, k=5)) -def initiate_new_sequence(args): +def initiate_new_sequence(args: argparse.Namespace) -> str: """ Generates a new sequence based on the parameters sent from the parser. @@ -148,15 +165,19 @@ def initiate_new_sequence(args): args.seed = get_rnd_seed() header = '' if not args.quiet: - header = (f'Seed: {args.seed}, Step: {args.step}, ' - f'Hash: {args.hash_algo}{os.linesep}') + header = ( + f'Seed: {args.seed}, Step: {args.step}, ' + f'Hash: {args.hash_algo}{os.linesep}' + ) generator = otp2289.generator.OTPGenerator( args.password.encode(), args.seed, - args.hash_algo) + args.hash_algo, + ) if args.challenge_string: return header + generator.generate_otp_hexdigest_from_challenge( - args.challenge_string) + args.challenge_string + ) return header + generator.generate_otp_hexdigest(args.step) @@ -164,92 +185,122 @@ def main(args=None): """the main entry point""" parser = argparse.ArgumentParser( prog=__package__, - epilog=(f'%(prog)s {otp2289.__version__} by Simeon Simeonov ' - '(sgs @ LiberaChat)'), - description='The following options are available') + epilog=( + f'%(prog)s {otp2289.__version__} by Simeon Simeonov ' + '(sgs @ LiberaChat)' + ), + description='The following options are available', + ) group = parser.add_mutually_exclusive_group(required=True) group.add_argument( '--generate-otp-range', action='store_true', dest='generate_otp_range', default=False, - help='Generates a range of OTP responses') + help='Generates a range of OTP responses', + ) group.add_argument( '--generate-otp-response', action='store_true', dest='generate_otp_response', default=False, - help='Generates a new OTP response') + help='Generates a new OTP response', + ) group.add_argument( '--initiate-new-sequence', action='store_true', dest='initiate_new_sequence', default=False, - help=('Initiates a new OTP sequence. Essentially the same as ' - '--generate-otp-response only it prompts twice for password ' - 'and always outputs hex (ignores -f).')) + help=( + 'Initiates a new OTP sequence. Essentially the same as ' + '--generate-otp-response only it prompts twice for password ' + 'and always outputs hex (ignores -f).' + ), + ) parser.add_argument( - '-a', '--hash-algorithm', + '-a', + '--hash-algorithm', metavar='', type=str, dest='hash_algo', default='md5', - help='The hash algorithm to use. Possible values: md5 (default), sha1') + help='The hash algorithm to use. Possible values: md5 (default), sha1', + ) parser.add_argument( - '-c', '--challenge-string', + '-c', + '--challenge-string', metavar='', type=str, dest='challenge_string', default='', - help='Use challenge string when generating response') + help='Use challenge string when generating response', + ) parser.add_argument( - '-f', '--output-format', + '-f', + '--output-format', metavar='', type=str, dest='output_format', default='hex', - help='The output format to use. Possible values: hex (default), token') + help='The output format to use. Possible values: hex (default), token', + ) parser.add_argument( - '-i', '--step', + '-i', + '--step', metavar='', type=int, dest='step', default=500, - help='The step. Default for initiating a new sequence is: 500') + help='The step. Default for initiating a new sequence is: 500', + ) parser.add_argument( - '-p', '--password', + '-p', + '--password', metavar='', type=str, dest='password', default='', - help=('The password or path to password file ' - '(default & recommended: prompt for passwd)')) + help=( + 'The password or path to password file ' + '(default & recommended: prompt for passwd)' + ), + ) parser.add_argument( - '-q', '--quiet', + '-q', + '--quiet', action='store_true', dest='quiet', default=False, - help='Dot not show headers. Only hex / tokens') + help='Dot not show headers. Only hex / tokens', + ) parser.add_argument( - '-r', '--range', + '-r', + '--range', metavar='', type=int, dest='range', default=1, - help='Amount of consecutive OTP hex/tokens to generate. default: 1') + help='Amount of consecutive OTP hex/tokens to generate. default: 1', + ) parser.add_argument( - '-s', '--seed', + '-s', + '--seed', metavar='[seed]', type=str, dest='seed', default='', - help=('The seed to use (1 to 16 alphanumeric characters) ' - '(default & recommended: random seed)')) + help=( + 'The seed to use (1 to 16 alphanumeric characters) ' + '(default & recommended: random seed)' + ), + ) parser.add_argument( - '-v', '--version', + '-v', + '--version', action='version', version=f'%(prog)s {otp2289.__version__}', - help='display program-version and exit') + help='display program-version and exit', + ) args = parser.parse_args(args) # handle the password before everything else if not args.password: @@ -257,8 +308,8 @@ def main(args=None): while True: args.password = getpass.getpass() if ( - not args.initiate_new_sequence or - args.password == getpass.getpass('Repeat password: ') + not args.initiate_new_sequence + or args.password == getpass.getpass('Repeat password: ') ): break eprint('The passwords do not match') @@ -267,7 +318,7 @@ def main(args=None): sys.exit(errno.EACCES) elif os.path.isfile(args.password): try: - with open(args.password, 'r') as fp: + with io.open(args.password, 'r', encoding='utf-8') as fp: args.password = fp.readline().strip() except Exception as exp: eprint(f'Unable to open password file: {exp}') diff --git a/otp2289/generator.py b/otp2289/generator.py index 5ea738b..7b86e1b 100644 --- a/otp2289/generator.py +++ b/otp2289/generator.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- # SPDX-License-Identifier: BSD-2-Clause-FreeBSD # -# Copyright (c) 2020, Simeon Simeonov +# Copyright (c) 2020-2022 Simeon Simeonov # All rights reserved. # # Redistribution and use in source and binary forms, with or without @@ -28,7 +28,6 @@ import binascii import hashlib import string - OTP_ALGO_MD5 = 1 OTP_ALGO_SHA1 = 2 @@ -306,7 +305,12 @@ class OTPChallengeException(Exception): class OTPGenerator: """OTPGenerator class""" - def __init__(self, password, seed='', hash_algo=OTP_ALGO_MD5): + def __init__( + self, + password: bytes, + seed: str = '', + hash_algo=OTP_ALGO_MD5, + ): """ Constructs an OTPGenerator object with a given password and seed. @@ -330,16 +334,19 @@ class OTPGenerator: raise OTPGeneratorException('Password must be a byte-string') if len(password) < 10: raise OTPGeneratorException( - 'Password must be longer than 10 bytes') + 'Password must be longer than 10 bytes' + ) self._password = password def __repr__(self): """repr implementation""" - return (f'{self.__class__} at {id(self)} (seed={self._seed}, ' - f'hash_algo={self._hash_algo})') + return ( + f'{self.__class__} at {id(self)} (seed={self._seed}, ' + f'hash_algo={self._hash_algo})' + ) @staticmethod - def bit_pair_sum(bit_stream): + def bit_pair_sum(bit_stream: str) -> int: """ Split bit_stream in bit-pairs and sum them all together. @@ -359,7 +366,7 @@ class OTPGenerator: return value @staticmethod - def bytes_to_tokens(hash_bytes): + def bytes_to_tokens(hash_bytes: bytes) -> str: """ Returns a 6 words token from bytes as specified by RFC-2289. @@ -369,8 +376,7 @@ class OTPGenerator: :return: 6 words tokens :rtype: str """ - bit_stream = ''.join( - ['{0:0>8b}'.format(byte) for byte in hash_bytes]) + bit_stream = ''.join([f'{byte:0>8b}' for byte in hash_bytes]) bit_pair_sum = OTPGenerator.bit_pair_sum(bit_stream) tokens = [] tokens.append(RFC1760_TOKENS[int(bit_stream[:11], 2)]) @@ -379,12 +385,17 @@ class OTPGenerator: tokens.append(RFC1760_TOKENS[int(bit_stream[33:44], 2)]) tokens.append(RFC1760_TOKENS[int(bit_stream[44:55], 2)]) tokens.append( - RFC1760_TOKENS[int( - bit_stream[55:64] + '{0:0>8b}'.format(bit_pair_sum)[-2:], 2)]) + RFC1760_TOKENS[ + int( + bit_stream[55:64] + f'{bit_pair_sum:0>8b}'[-2:], + 2, + ) + ] + ) return ' '.join(tokens) @staticmethod - def get_tokens_from_challenge(challenge): + def get_tokens_from_challenge(challenge: str) -> tuple: """ Returns tokens (seed, hash_algo and step) from a challenge string. @@ -410,7 +421,7 @@ class OTPGenerator: raise OTPChallengeException('Invalid challenge') from None @staticmethod - def sha1_digest_folding(sha1_digest): + def sha1_digest_folding(sha1_digest: bytes) -> bytes: """ Implementation of the 160bit -> 64bit folding algorithm for sha1 digest. @@ -425,14 +436,17 @@ class OTPGenerator: raise OTPGeneratorException('sha1_digest must be of type bytes') if len(sha1_digest) != 20: raise OTPGeneratorException( - 'sha1_digest must be 160 bits (20 bytes) long') + 'sha1_digest must be 160 bits (20 bytes) long' + ) digested = list(5 * b'i') # 5 bytes (40 bits) result = list(8 * b'x') # 8 bytes (64 bits) for i in range(5): - digested[i] = (((sha1_digest[i * 4 + 0] & 0xff) << 24) | - ((sha1_digest[i * 4 + 1] & 0xff) << 16) | - ((sha1_digest[i * 4 + 2] & 0xff) << 8) | - (sha1_digest[i * 4 + 3] & 0xff)) + digested[i] = ( + ((sha1_digest[i * 4 + 0] & 0xFF) << 24) + | ((sha1_digest[i * 4 + 1] & 0xFF) << 16) + | ((sha1_digest[i * 4 + 2] & 0xFF) << 8) + | (sha1_digest[i * 4 + 3] & 0xFF) + ) # sha.digest[0] ^= sha.digest[2]; # sha.digest[1] ^= sha.digest[3]; # sha.digest[0] ^= sha.digest[4]; @@ -446,18 +460,18 @@ class OTPGenerator: # result[j+3] = (unsigned char)((sha.digest[i] >> 24) & 0xff); # } # just hardcoding the two iterations for better efficiency - result[0] = digested[0] & 0xff - result[1] = (digested[0] >> 8) & 0xff - result[2] = (digested[0] >> 16) & 0xff - result[3] = (digested[0] >> 24) & 0xff - result[4] = digested[1] & 0xff - result[5] = (digested[1] >> 8) & 0xff - result[6] = (digested[1] >> 16) & 0xff - result[7] = (digested[1] >> 24) & 0xff + result[0] = digested[0] & 0xFF + result[1] = (digested[0] >> 8) & 0xFF + result[2] = (digested[0] >> 16) & 0xFF + result[3] = (digested[0] >> 24) & 0xFF + result[4] = digested[1] & 0xFF + result[5] = (digested[1] >> 8) & 0xFF + result[6] = (digested[1] >> 16) & 0xFF + result[7] = (digested[1] >> 24) & 0xFF return bytes(result) @staticmethod - def strxor(byte_str1, byte_str2): + def strxor(byte_str1: bytes, byte_str2: bytes) -> bytes: """ Implementation of strxor similar to the one provided by pycrypto. @@ -472,17 +486,17 @@ class OTPGenerator: """ if not (isinstance(byte_str1, bytes) and isinstance(byte_str2, bytes)): raise OTPGeneratorException( - 'byte_str1 and byte_str2 must be of type bytes') + 'byte_str1 and byte_str2 must be of type bytes' + ) length = len(byte_str1) if length != len(byte_str2) or length < 1: raise OTPGeneratorException( - 'byte_str1 and byte_str2 must be of the same length > 0') - return bytes( - [byte_str1[i] ^ byte_str2[i] for i in range(length)] - ) + 'byte_str1 and byte_str2 must be of the same length > 0' + ) + return bytes([byte_str1[i] ^ byte_str2[i] for i in range(length)]) @staticmethod - def tokens_to_bytes(tokens_str): + def tokens_to_bytes(tokens_str: str) -> bytes: """ Returns bytes from a 6 words token as specified by RFC-2289. @@ -499,14 +513,17 @@ class OTPGenerator: tokens = tokens_str.split() if len(tokens) != 6: raise OTPGeneratorException( - 'Tokens-string does not contain 6 tokens') + 'Tokens-string does not contain 6 tokens' + ) token_ints = [] try: - token_ints = [RFC1760_TOKENS.index(token.upper()) for token in - tokens] + token_ints = [ + RFC1760_TOKENS.index(token.upper()) for token in tokens + ] except ValueError: raise OTPGeneratorException( - 'One or more words not present in RFC1760') from None + 'One or more words not present in RFC1760' + ) from None # now we build a string of bits bit_stream = format(token_ints[0], '011b') bit_stream += format(token_ints[1], '011b') @@ -519,14 +536,14 @@ class OTPGenerator: # OTP servers MUST verify this checksum explicitly as part of the # operation of decoding this representation of the one-time password. if ( - '{0:0>8b}'.format(OTPGenerator.bit_pair_sum( - bit_stream[:64]))[-2:] != bit_stream[-2:] + f'{OTPGenerator.bit_pair_sum(bit_stream[:64]):0>8b}'[-2:] + != bit_stream[-2:] ): raise OTPGeneratorException('Invalid bit checksum') return int(bit_stream[:64], 2).to_bytes(8, 'big') @staticmethod - def validate_hash_algo(hash_algo): + def validate_hash_algo(hash_algo) -> str: """ Validates the provided hash-algorithm. @@ -541,19 +558,20 @@ class OTPGenerator: if isinstance(hash_algo, int): if hash_algo not in _ALGO_DICT: raise OTPGeneratorException( - 'hash_algo is not among the known algorithms') + 'hash_algo is not among the known algorithms' + ) hash_algo = _ALGO_DICT.get(hash_algo) if not isinstance(hash_algo, str): - raise OTPGeneratorException( - 'hash_algo must be an int or a str') + raise OTPGeneratorException('hash_algo must be an int or a str') if hash_algo not in hashlib.algorithms_available: raise OTPGeneratorException( f'{hash_algo} is not supported by this version of the ' - 'hashlib module') + 'hashlib module' + ) return hash_algo @staticmethod - def validate_seed(seed): + def validate_seed(seed: str) -> str: """ Validates the provided seed as defined by RFC-2289. @@ -569,15 +587,17 @@ class OTPGenerator: raise OTPGeneratorException('Seed must be a string') if not seed or len(seed) > 16: raise OTPGeneratorException( - 'The seed MUST be of 1 to 16 characters in length') + 'The seed MUST be of 1 to 16 characters in length' + ) for char in seed: if char not in string.ascii_letters + string.digits: raise OTPGeneratorException( - 'The seed MUST consist of purely alphanumeric characters') + 'The seed MUST consist of purely alphanumeric characters' + ) return seed @staticmethod - def validate_step(step): + def validate_step(step: int) -> int: """ Validates the provided step as defined by RFC-2289. @@ -595,7 +615,7 @@ class OTPGenerator: raise OTPGeneratorException('Step value MUST be >= 0') return step - def generate_otp_hexdigest(self, step): + def generate_otp_hexdigest(self, step: int) -> str: """ Generates the OTP hexdigest for the given step. @@ -607,7 +627,7 @@ class OTPGenerator: """ return '0x' + binascii.hexlify(self._generate_otp_bytes(step)).decode() - def generate_otp_hexdigest_from_challenge(self, challenge): + def generate_otp_hexdigest_from_challenge(self, challenge: str) -> str: """ Same as generate_otp_hexdigest, but it generates hex. from a challenge. @@ -628,7 +648,7 @@ class OTPGenerator: self._hash_algo = self.validate_hash_algo(hash_algo) return self.generate_otp_hexdigest(step) - def generate_otp_words(self, step): + def generate_otp_words(self, step: int) -> str: """ Generates the OTP six words token for the given step. @@ -640,7 +660,7 @@ class OTPGenerator: """ return self.bytes_to_tokens(self._generate_otp_bytes(step)) - def generate_otp_words_from_challenge(self, challenge): + def generate_otp_words_from_challenge(self, challenge: str) -> str: """ Same as generate_otp_words, but it generates words from a challenge. @@ -661,7 +681,7 @@ class OTPGenerator: self._hash_algo = self.validate_hash_algo(hash_algo) return self.generate_otp_words(step) - def hexdigest_range(self, start=499, stop=0): + def hexdigest_range(self, start: int = 499, stop: int = 0): """ Returns an iterator that providing hexdigests corresponding to steps from `start` to and including `stop`. @@ -679,11 +699,12 @@ class OTPGenerator: raise OTPGeneratorException('Step value MUST be an int') if start < stop: raise OTPGeneratorException( - 'Start value can not be lower than stop') + 'Start value can not be lower than stop' + ) for step in range(start, stop - 1, -1): yield self.generate_otp_hexdigest(step) - def words_range(self, start=499, stop=0): + def words_range(self, start: int = 499, stop: int = 0): """ Returns an iterator that providing the words corresponding to steps from `start` to and including `stop`. @@ -701,11 +722,12 @@ class OTPGenerator: raise OTPGeneratorException('Step value MUST be an int') if start < stop: raise OTPGeneratorException( - 'Start value can not be lower than stop') + 'Start value can not be lower than stop' + ) for step in range(start, stop - 1, -1): yield self.generate_otp_words(step) - def _generate_otp_bytes(self, step): + def _generate_otp_bytes(self, step: int) -> bytes: """ Generates the OTP bytes for the given step. @@ -733,6 +755,6 @@ class OTPGenerator: digest = self.sha1_digest_folding(large_digest) else: raise OTPGeneratorException( - '{hash_algo} is not supported by this module'.format( - hash_algo=self._hash_algo)) + f'{self._hash_algo} is not supported by this module' + ) return digest diff --git a/otp2289/server.py b/otp2289/server.py index d0c3e01..7f742cd 100644 --- a/otp2289/server.py +++ b/otp2289/server.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- # SPDX-License-Identifier: BSD-2-Clause-FreeBSD # -# Copyright (c) 2020, Simeon Simeonov +# Copyright (c) 2020-2022 Simeon Simeonov # All rights reserved. # # Redistribution and use in source and binary forms, with or without @@ -27,9 +27,7 @@ import binascii import hashlib -from .generator import (OTP_ALGO_MD5, - OTPGenerator, - OTPGeneratorException) +from .generator import OTP_ALGO_MD5, OTPGenerator, OTPGeneratorException class OTPStateException(Exception): @@ -53,7 +51,13 @@ class OTPState: - validate the corresponding generated response from the generator """ - def __init__(self, ot_hex, current_step, seed, hash_algo=OTP_ALGO_MD5): + def __init__( + self, + ot_hex: str, + current_step: int, + seed: str, + hash_algo=OTP_ALGO_MD5, + ): """ Constructs an OTPState object with the given arguments. @@ -87,45 +91,47 @@ class OTPState: 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})') + 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): + 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): + def current_digest(self) -> bytes: """current_digest-property""" return self._current_digest @property - def hash_algo(self): + def hash_algo(self) -> str: """hash_algo-property""" return self._hash_algo @property - def seed(self): + def seed(self) -> str: """seed-property""" return self._seed @property - def step(self): + def step(self) -> int: """step-property""" return self._step @property - def validated(self): + def validated(self) -> bool: """validated-property""" return bool(self._new_digest_hex) @classmethod - def from_dict(cls, dict_obj): + def from_dict(cls, dict_obj: dict): """ Returns an OTPState object from the dict-object @@ -138,7 +144,7 @@ class OTPState: return cls(**dict_obj) @staticmethod - def response_to_bytes(response): + def response_to_bytes(response: str) -> bytes: """ A wrapper that handles/validates the response as specified by RFC-2289. @@ -166,10 +172,11 @@ class OTPState: return OTPState.validate_hex(response) except OTPStateException: raise OTPInvalidResponse( - 'The response is neither a valid token or hex') from None + 'The response is neither a valid token or hex' + ) from None @staticmethod - def validate_hex(ot_hex): + def validate_hex(ot_hex: str) -> bytes: """ Validates the provided hexidigest. @@ -187,8 +194,10 @@ class OTPState: 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)') + raise OTPStateException( + 'The length of the hex should be 16 ' + '(representing 64 bits digest)' + ) try: return binascii.unhexlify(ot_hex) except binascii.Error: @@ -206,12 +215,18 @@ class OTPState: """ 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, store_valid_response=True): + 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. @@ -233,32 +248,35 @@ class OTPState: 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 + 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() + 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 + 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() + 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): + def to_dict(self) -> dict: """ Returns a dict representation of the object. @@ -270,10 +288,12 @@ class OTPState: 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} + return { + 'ot_hex': ot_hex, + 'current_step': self._step, + 'seed': self._seed, + 'hash_algo': self._hash_algo, + } class OTPStore: @@ -285,6 +305,7 @@ class OTPStore: The class could serve as a base class when implementing store backends. """ + def __init__(self, data=None): """ Constructs an OTPStore object from data @@ -310,7 +331,7 @@ class OTPStore: return len(self._data) @property - def data(self): + def data(self) -> dict: """ data-property @@ -320,7 +341,7 @@ class OTPStore: return self._data @property - def states(self): + def states(self) -> dict: """ states-property @@ -329,7 +350,7 @@ class OTPStore: """ return self._states - def add_state(self, key, state): + def add_state(self, key: str, state: OTPState): """ Adds an OTPState object with a given key. @@ -356,7 +377,7 @@ class OTPStore: """A wrapper for dict.items""" return self._data.items() - def pop_state(self, key): + def pop_state(self, key: str) -> OTPState: """ Removes specified key and returns the corresponding OTPState-object. @@ -376,7 +397,12 @@ class OTPStore: self._states.pop(state) return state - def response_validates(self, key, response, store_valid_response=True): + 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. @@ -411,7 +437,7 @@ class OTPStore: self._states.pop(state) return rvalue - def to_dict(self): + def to_dict(self) -> dict: """ Returns a dict representation of the object. @@ -422,7 +448,7 @@ class OTPStore: """ return {key: state.to_dict() for key, state in self._data.items()} - def _add_data(self, dict_obj): + def _add_data(self, dict_obj: dict) -> dict: """ Adds data from a dict object (dict_obj). -- cgit v1.3