From e297c0696cf594f8f700dd82fdcdda582e348eda Mon Sep 17 00:00:00 2001 From: Simeon Simeonov Date: Mon, 6 Apr 2020 10:28:46 +0200 Subject: Add a simple CLI interface with tests and update README.md --- README.md | 97 +++++++++++++++-- otp2289/__main__.py | 293 ++++++++++++++++++++++++++++++++++++++++++++++++++++ otp2289/server.py | 18 ++-- tests/test_main.py | 132 +++++++++++++++++++++++ 4 files changed, 525 insertions(+), 15 deletions(-) create mode 100644 otp2289/__main__.py create mode 100644 tests/test_main.py diff --git a/README.md b/README.md index ac3bca9..5811b54 100644 --- a/README.md +++ b/README.md @@ -70,20 +70,99 @@ one-time password." ## Examples - ```python - import getpass - - import otp2289 +We define the two entities: *client* and *server*. The entire application of +RFC-2289 consists of interactions between them. - # create a generator object - passwd_bytes = getpass.getpass().encode() # Type: This is a test. + ```python + # + import getpass # client only + + import otp2289 # client and server + + # the server starts by picking: + # - algorithm (MD5 or SHA1) to use + # - seed - 1 to 16 alphanumeric characters. The seed must never be reused. + # - initial step - number (int) that will be decremented for each OTP. + # In FreeBSD, the following default values are used: + # - MD5 + # - the first two letters of the hostname + 5 random digits for seed + # - initial step: 500 + + # the client receives those values, chooses a strong password and creates + # initialization digest (hash). The password 'This is a test.' will give you + # the same results as in the following example. + passwd_bytes = getpass.getpass().encode() # Fetch the password as bytes generator = otp2289.generator.OTPGenerator(passwd_bytes, 'TesT', otp2289.OTP_ALGO_MD5) - generator.generate_otp_hexdigest(0) - generator.gen.generate_otp_words(0) + digest = generator.generate_otp_hexdigest(500) + # digest is now: 0x2b8d82b6ac14346c + # the client sends it to the server + + # the server creates the first state. Note that step is decremented by 1: + state = otp2289.server.OTPState(digest, 499, 'TesT', otp2289.OTP_ALGO_MD5) + # the state can be stored in a OTPStore container: + store = otp2289.server.OTPStore() + # key can be any str that can be used to reference the state (f.i username) + store.add_state('myusername', state) # where key can be any str that can be + # OTPStore is provided only for convenience as it is not part of RFC-2289. + # The server can store states any way it wants. A normal dict is also fine. + # Once the initial state is set on the server, the client can authenticate. + + # Upon authentication request (f.i. login), the server issues a challenge + # based on the state: + challenge = state.challenge_string # challenge is now 'otp-md5 499 TesT ' + + # the client can now respond by using (or recreating) the same generator + # created earlier. RFC-2289 defines two types of responses: + # - hex (like '0x2b8d82b6ac14346c') - more suited for automation + # - tokens consisting of 6 short words - better when responding manually + hex_response = generator.generate_otp_hexdigest(499) # '0x6323f96296a2526b' + token_response = generator.generate_otp_words(499) + # token_response is now: 'CANT JAW BITS NU LO PUP' + # a possible shortcut may be to use the challenge-string directly: + hex_response = generator.generate_otp_hexdigest_from_challenge(challenge) + token_response = generator.generate_otp_words_from_challenge(challenge) + # ... giving the same results. + + # once the response is received, the server validates it by yet again using + # the current state: + result = state.response_validates(hex_response) + # or + result = state.response_validates(token_response) + # result should be True if the response matches the state, False if not + # in case of invalid response or response checksum doesn't match, a + # otp2289.server.OTPInvalidResponse exception is raised. + + # once the state has successfully validated the corresponding response, + the state **must never be used again** and a state corresponding to the + "next" (498) step created. + state = state.get_next_state() + + # the next authentication attempt... + challenge = state.challenge_string # challenge is now 'otp-md5 498 TesT ' + # ... and on the client side... + hex_response = generator.generate_otp_hexdigest_from_challenge(challenge) + # etc. etc... + ``` + +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: + + ```bash + python -m otp2289 --generate-otp-response -f token -i 498 -s TesT ``` +... will prompt for password and generate a 6 words (token) response. + + ```bash + python -m otp2289 --generate-otp-range -f token -i 498 -s TesT + ``` + +... will prompt for password and generate a range of 4 one-time passwords +starting from (and including) 498. + ## Author @@ -95,5 +174,5 @@ Simeon Simeonov - sgs @ Freenode Copyright (c) 2020, Simeon Simeonov All rights reserved. -[licensed](LICENSE) under the BSD 2-clause. +[Licensed](LICENSE) under the BSD 2-clause. SPDX-License-Identifier: BSD-2-Clause-FreeBSD diff --git a/otp2289/__main__.py b/otp2289/__main__.py new file mode 100644 index 0000000..2962d29 --- /dev/null +++ b/otp2289/__main__.py @@ -0,0 +1,293 @@ +# -*- coding: utf-8 -*- +# SPDX-License-Identifier: BSD-2-Clause-FreeBSD +# +# Copyright (c) 2020, 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. +""" +CLI entry point for the otp2289 package + +Examples: +python -m otp2289 --initiate-new-sequence -s TesT + +python -m otp2289 --generate-otp-response -c "otp-md5 499 TesT " -f token +python -m otp2289 --generate-otp-response -s TesT -i 499 -f token +""" +import argparse +import errno +import getpass +import os +import secrets +import string +import sys + +import otp2289 + + +def eprint(*arg, **kwargs): + """stdderr print wrapper""" + print(*arg, file=sys.stderr, flush=True, **kwargs) + + +def generate_otp_response(args): + """ + Generates a response based on the parameters sent from the parser + + :param args: The arguments assigned from argparse + :type args: argparse.Namespace + + :raises OTPChallengeException: In case of invalid challenge + + :raises OTPGeneratorException: In case of wrong generator parameters + + :return: The response string + :rtype: str + """ + generator = otp2289.generator.OTPGenerator( + args.password.encode(), + args.seed, + args.hash_algo) + if args.challenge_string: + if args.output_format == 'token': + return generator.generate_otp_words_from_challenge( + args.challenge_string) + return generator.generate_otp_hexdigest_from_challenge( + 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}') + 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): + """ + Generates range of responses based on the parameters sent from the parser + + :param args: The arguments assigned from argparse + :type args: argparse.Namespace + + :raises OTPChallengeException: In case of invalid challenge + + :raises OTPGeneratorException: In case of wrong generator parameters + + :return: The responses string + :rtype: str + """ + generator = otp2289.generator.OTPGenerator( + args.password.encode(), + args.seed, + args.hash_algo) + if args.output_format == 'token': + method = generator.generate_otp_words + else: + method = generator.generate_otp_hexdigest + # handle most cases explicitly + if args.range == 1: + return f'{args.step}: ' + method(args.step) + if args.range > args.step + 1: + args.range = args.step + 1 + # 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}') + return header + os.linesep.join( + [f'{step}: ' + method(step) for step in range( + args.step, args.step - args.range, -1)]) + + +def get_rnd_seed(): + """ + 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))) + + +def initiate_new_sequence(args): + """ + Generates a new sequence based on the parameters sent from the parser. + + :param args: The arguments assigned from argparse + :type args: argparse.Namespace + + :raises OTPChallengeException: In case of invalid challenge + + :raises OTPGeneratorException: In case of wrong generator parameters + + :return: The response string + :rtype: str + """ + if not args.seed: + 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}') + generator = otp2289.generator.OTPGenerator( + args.password.encode(), + args.seed, + args.hash_algo) + if args.challenge_string: + return header + generator.generate_otp_hexdigest_from_challenge( + args.challenge_string) + return header + generator.generate_otp_hexdigest(args.step) + + +def main(args=None): + """the main entry point""" + parser = argparse.ArgumentParser( + prog=__package__, + epilog=(f'%(prog)s {otp2289.__version__} by Simeon Simeonov ' + '(sgs @ Freenode)'), + 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') + group.add_argument( + '--generate-otp-response', + action='store_true', + dest='generate_otp_response', + default=False, + 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).')) + parser.add_argument( + '-a', '--hash-algorithm', + metavar='', + type=str, + dest='hash_algo', + default='md5', + help='The hash algorithm to use. Possible values: md5 (default), sha1') + parser.add_argument( + '-c', '--challenge-string', + metavar='', + type=str, + dest='challenge_string', + default='', + help='Use challenge string when generating response') + parser.add_argument( + '-f', '--output-format', + metavar='', + type=str, + dest='output_format', + default='hex', + help='The output format to use. Possible values: hex (default), token') + parser.add_argument( + '-i', '--step', + metavar='', + type=int, + dest='step', + default=500, + help='The step. Default for initiating a new sequence is: 500') + parser.add_argument( + '-p', '--password', + metavar='', + type=str, + dest='password', + default='', + help=('The password or path to password file ' + '(default & recommended: prompt for passwd)')) + parser.add_argument( + '-q', '--quiet', + action='store_true', + dest='quiet', + default=False, + help='Dot not show headers. Only hex / tokens') + parser.add_argument( + '-r', '--range', + metavar='', + type=int, + dest='range', + default=1, + help='Amount of consecutive OTP hex/tokens to generate. default: 1') + parser.add_argument( + '-s', '--seed', + metavar='[seed]', + type=str, + dest='seed', + default='', + help=('The seed to use (1 to 16 alphanumeric characters) ' + '(default & recommended: random seed)')) + parser.add_argument( + '-v', '--version', + action='version', + version=f'%(prog)s {otp2289.__version__}', + help='display program-version and exit') + args = parser.parse_args(args) + # handle the password before everything else + if not args.password: + try: + while True: + args.password = getpass.getpass() + if ( + not args.initiate_new_sequence or + args.password == getpass.getpass('Repeat password: ') + ): + break + eprint('The passwords do not match') + except KeyboardInterrupt: + eprint(os.linesep + 'Prompt terminated') + sys.exit(errno.EACCES) + elif os.path.isfile(args.password): + try: + with open(args.password, 'r') as fp: + args.password = fp.readline().strip() + except Exception as exp: + eprint(f'Unable to open password file: {exp}') + sys.exit(1) + try: + if args.initiate_new_sequence: + print(initiate_new_sequence(args)) + if args.generate_otp_range: + print(generate_otp_range(args)) + if args.generate_otp_response: + print(generate_otp_response(args)) + sys.exit(0) + except otp2289.generator.OTPGeneratorException as exp: + eprint(f'GeneratorException: {exp}') + except otp2289.generator.OTPChallengeException as exp: + eprint(f'ChallengeException: {exp}') + except Exception as exp: + eprint(f'Unknown error: {exp}') + sys.exit(1) + + +if __name__ == '__main__': + main() diff --git a/otp2289/server.py b/otp2289/server.py index b765af6..fca3f06 100644 --- a/otp2289/server.py +++ b/otp2289/server.py @@ -58,10 +58,9 @@ class OTPState: Constructs an OTPState object with the given arguments. Keyword Arguments: - :param ot_hex: The one-time hex from the last successful - authentication or the first OTP of a newly - initialized sequence - :type ot_hex: str + :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 @@ -81,7 +80,9 @@ class OTPState: self._step = OTPGenerator.validate_step(current_step) except OTPGeneratorException as exp: raise OTPStateException(exp.args[0]) - self._current_digest = self.validate_hex(ot_hex) + 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 @property @@ -223,6 +224,7 @@ 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 ): @@ -234,6 +236,7 @@ class OTPState: 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 @@ -255,7 +258,10 @@ class OTPState: :return: The dict representation of the object :rtype: dict """ - return {'ot_hex': binascii.hexlify(self._current_digest).decode(), + 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} diff --git a/tests/test_main.py b/tests/test_main.py new file mode 100644 index 0000000..b48f625 --- /dev/null +++ b/tests/test_main.py @@ -0,0 +1,132 @@ +# -*- coding: utf-8 -*- +# SPDX-License-Identifier: BSD-2-Clause-FreeBSD +# +# Copyright (c) 2020, 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. +"""Tests for otp2289.__main__""" +import os + +import pytest + +from otp2289.__main__ import main + + +def test_main_generate_otp_response(capsys): + """tests main""" + args = ['--generate-otp-response', + '-a', + 'sha1', + '-i', + '99', + '-s', + 'TesT', + '-p', + 'This is a test.'] + with pytest.raises(SystemExit) as exit_info: + main(args) + captured = capsys.readouterr() + assert captured.out == (f'Seed: TesT, Step: 99, Hash: sha1{os.linesep}' + f'0x87fec7768b73ccf9{os.linesep}') + assert exit_info.type == SystemExit + assert exit_info.value.code == 0 + args.extend(['-f', 'token']) + with pytest.raises(SystemExit) as exit_info: + main(args) + captured = capsys.readouterr() + assert captured.out == (f'Seed: TesT, Step: 99, Hash: sha1{os.linesep}' + f'GAFF WAIT SKID GIG SKY EYED{os.linesep}') + assert exit_info.type == SystemExit + assert exit_info.value.code == 0 + args.append('-q') + with pytest.raises(SystemExit) as exit_info: + main(args) + captured = capsys.readouterr() + assert captured.out == f'GAFF WAIT SKID GIG SKY EYED{os.linesep}' + assert exit_info.type == SystemExit + assert exit_info.value.code == 0 + + +def test_main_generate_otp_range(capsys): + """tests main""" + args = ['--generate-otp-range', + '-i', + '2', + '-s', + 'TesT', + '-r', + '5', + '-p', + 'This is a test.'] + with pytest.raises(SystemExit) as exit_info: + main(args) + captured = capsys.readouterr() + assert captured.out == (f'Seed: TesT, Step: 2, Hash: md5, Range: 3' + f'{os.linesep}' + f'2: 0x4049f8b161669b7b{os.linesep}' + f'1: 0x7965e05436f5029f{os.linesep}' + f'0: 0x9e876134d90499dd{os.linesep}') + assert exit_info.type == SystemExit + assert exit_info.value.code == 0 + args.append('-q') + with pytest.raises(SystemExit) as exit_info: + main(args) + captured = capsys.readouterr() + assert captured.out == (f'2: 0x4049f8b161669b7b{os.linesep}' + f'1: 0x7965e05436f5029f{os.linesep}' + f'0: 0x9e876134d90499dd{os.linesep}') + assert exit_info.type == SystemExit + assert exit_info.value.code == 0 + args.extend(['-f', 'token']) + with pytest.raises(SystemExit) as exit_info: + main(args) + captured = capsys.readouterr() + assert captured.out == (f'2: THY AVON NO NECK COKE MOLL{os.linesep}' + f'1: EASE OIL FUM CURE AWRY AVIS{os.linesep}' + f'0: INCH SEA ANNE LONG AHEM TOUR{os.linesep}') + assert exit_info.type == SystemExit + assert exit_info.value.code == 0 + + +def test_main_initiate(capsys): + """tests main""" + args = ['--initiate-new-sequence', + '-i', + '500', + '-s', + 'TesT', + '-p', + 'This is a test.'] + with pytest.raises(SystemExit) as exit_info: + main(args) + captured = capsys.readouterr() + assert captured.out == (f'Seed: TesT, Step: 500, Hash: md5{os.linesep}' + f'0x2b8d82b6ac14346c{os.linesep}') + assert exit_info.type == SystemExit + assert exit_info.value.code == 0 + args.append('-q') + with pytest.raises(SystemExit) as exit_info: + main(args) + captured = capsys.readouterr() + assert captured.out == f'0x2b8d82b6ac14346c{os.linesep}' + assert exit_info.type == SystemExit + assert exit_info.value.code == 0 -- cgit v1.3