suppor writing csv and json to stdout; print errors to stderr

- add tests

README.md

@@ -38,9 +38,9 @@ positional arguments:
 
 options:
   -h, --help               show this help message and exit
-  --json FILE, -j FILE     export json file
+  --json FILE, -j FILE     export json file or - for stdout
-  --csv FILE, -c FILE      export csv file
+  --csv FILE, -c FILE      export csv file or - for stdout
-  --keepass FILE, -k FILE  export totp/hotp csv file(s) for KeePass
+  --keepass FILE, -k FILE  export totp/hotp csv file(s) for KeePass, - for stdout
   --printqr, -p            print QR code(s) as text to the terminal (requires qrcode module)
   --saveqr DIR, -s DIR     save QR code(s) as images to the given folder (requires qrcode module)
   --verbose, -v            verbose output
@@ -75,6 +75,8 @@ For printing QR codes, the qrcode module is required, otherwise it can be omitte
     * JSON
     * Dedicated CSV for KeePass
     * QR code images
+* Supports reading from stdin and writing to stdout by specifying '-'
+* Errors and warnings are written to stderr
 * Many ways to run the script:
     * Native Python
     * pipenv

extract_otp_secret_keys.py

@@ -59,6 +59,10 @@ def sys_main():
 
 def main(sys_args):
     global verbose, quiet
+
+    # allow to use sys.stdout with with (avoid closing)
+    sys.stdout.close = lambda: None
+
     args = parse_args(sys_args)
     verbose = args.verbose if args.verbose else 0
     quiet = args.quiet
@@ -73,16 +77,16 @@ def parse_args(sys_args):
     formatter = lambda prog: argparse.HelpFormatter(prog,max_help_position=52)
     arg_parser = argparse.ArgumentParser(formatter_class=formatter)
     arg_parser.add_argument('infile', help='file or - for stdin (default: -) with "otpauth-migration://..." URLs separated by newlines, lines starting with # are ignored')
-    arg_parser.add_argument('--json', '-j', help='export json file', metavar=('FILE'))
+    arg_parser.add_argument('--json', '-j', help='export json file or - for stdout', metavar=('FILE'))
-    arg_parser.add_argument('--csv', '-c', help='export csv file', metavar=('FILE'))
+    arg_parser.add_argument('--csv', '-c', help='export csv file or - for stdout', metavar=('FILE'))
-    arg_parser.add_argument('--keepass', '-k', help='export totp/hotp csv file(s) for KeePass', metavar=('FILE'))
+    arg_parser.add_argument('--keepass', '-k', help='export totp/hotp csv file(s) for KeePass, - for stdout', metavar=('FILE'))
     arg_parser.add_argument('--printqr', '-p', help='print QR code(s) as text to the terminal (requires qrcode module)', action='store_true')
     arg_parser.add_argument('--saveqr', '-s', help='save QR code(s) as images to the given folder (requires qrcode module)', metavar=('DIR'))
     arg_parser.add_argument('--verbose', '-v', help='verbose output', action='count')
     arg_parser.add_argument('--quiet', '-q', help='no stdout output', action='store_true')
     args = arg_parser.parse_args(sys_args)
     if args.verbose and args.quiet:
-        print("The arguments --verbose and --quiet are mutually exclusive.")
+        eprint("The arguments --verbose and --quiet are mutually exclusive.")
         sys.exit(1)
     return args
 
@@ -136,7 +140,7 @@ def extract_otps(args):
 def get_payload_from_line(line, i, args):
     global verbose
     if not line.startswith('otpauth-migration://'):
-        print('\nWARN: line is not a otpauth-migration:// URL\ninput file: {}\nline "{}"\nProbably a wrong file was given'.format(args.infile, line))
+        eprint('\nWARN: line is not a otpauth-migration:// URL\ninput file: {}\nline "{}"\nProbably a wrong file was given'.format(args.infile, line))
     parsed_url = urlparse(line)
     if verbose > 1: print('\nDEBUG: parsed_url={}'.format(parsed_url))
     try:
@@ -145,7 +149,7 @@ def get_payload_from_line(line, i, args):
         params = []
     if verbose > 1: print('\nDEBUG: querystring params={}'.format(params))
     if 'data' not in params:
-        print('\nERROR: no data query parameter in input URL\ninput file: {}\nline "{}"\nProbably a wrong file was given'.format(args.infile, line))
+        eprint('\nERROR: no data query parameter in input URL\ninput file: {}\nline "{}"\nProbably a wrong file was given'.format(args.infile, line))
         sys.exit(1)
     data_base64 = params['data'][0]
     if verbose > 1: print('\nDEBUG: data_base64={}'.format(data_base64))
@@ -156,8 +160,8 @@ def get_payload_from_line(line, i, args):
     try:
         payload.ParseFromString(data)
     except:
-        print('\nERROR: Cannot decode otpauth-migration migration payload.')
+        eprint('\nERROR: Cannot decode otpauth-migration migration payload.')
-        print('data={}'.format(data_base64))
+        eprint('data={}'.format(data_base64))
         exit(1);
     if verbose:
         print('\n{}. Payload Line'.format(i), payload, sep='\n')
@@ -228,7 +232,7 @@ def print_qr(args, data):
 def write_csv(args, otps):
     global verbose, quiet
     if args.csv and len(otps) > 0:
-        with open(args.csv, "w") as outfile:
+        with open_file_or_stdout_for_csv(args.csv) as outfile:
             writer = csv.DictWriter(outfile, otps[0].keys())
             writer.writeheader()
             writer.writerows(otps)
@@ -245,7 +249,7 @@ def write_keepass_csv(args, otps):
         count_totp_entries = 0
         count_hotp_entries = 0
         if has_totp:
-            with open(otp_filename_totp, "w") as outfile:
+            with open_file_or_stdout_for_csv(otp_filename_totp) as outfile:
                 writer = csv.DictWriter(outfile, ["Title", "User Name", "TimeOtp-Secret-Base32", "Group"])
                 writer.writeheader()
                 for otp in otps:
@@ -258,7 +262,7 @@ def write_keepass_csv(args, otps):
                         })
                         count_totp_entries += 1
         if has_hotp:
-            with open(otp_filename_hotp, "w") as outfile:
+            with open_file_or_stdout_for_csv(otp_filename_hotp) as outfile:
                 writer = csv.DictWriter(outfile, ["Title", "User Name", "HmacOtp-Secret-Base32", "HmacOtp-Counter", "Group"])
                 writer.writeheader()
                 for otp in otps:
@@ -279,7 +283,7 @@ def write_keepass_csv(args, otps):
 def write_json(args, otps):
     global verbose, quiet
     if args.json:
-        with open(args.json, "w") as outfile:
+        with open_file_or_stdout(args.json) as outfile:
             json.dump(otps, outfile, indent=4)
         if not quiet: print("Exported {} otp entries to json {}".format(len(otps), args.json))
 
@@ -297,5 +301,25 @@ def add_pre_suffix(file, pre_suffix):
     return name + "." + pre_suffix + (ext if ext else "")
 
 
+def open_file_or_stdout(filename):
+    '''stdout is denoted as "-".
+    Note: Set before the following line:
+    sys.stdout.close = lambda: None'''
+    return open(filename, "w") if filename != '-' else sys.stdout
+
+
+def open_file_or_stdout_for_csv(filename):
+    '''stdout is denoted as "-".
+    newline=''
+    Note: Set before the following line:
+    sys.stdout.close = lambda: None'''
+    return open(filename, "w", newline='') if filename != '-' else sys.stdout
+
+
+def eprint(*args, **kwargs):
+    '''Print to stderr.'''
+    print(*args, file=sys.stderr, **kwargs)
+
+
 if __name__ == '__main__':
     sys_main()

test_extract_otp_secret_keys_pytest.py

@@ -18,7 +18,7 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
-from utils import read_csv, read_json, remove_files, remove_dir_with_files, read_file_to_str, file_exits
+from utils import read_csv, read_csv_str, read_json, read_json_str, remove_files, remove_dir_with_files, read_file_to_str, file_exits
 from os import path
 from pytest import raises
 from io import StringIO
@@ -73,6 +73,28 @@ def test_extract_csv(capsys):
     cleanup()
 
 
+def test_extract_csv_stdout(capsys):
+    # Arrange
+    cleanup()
+
+    # Act
+    extract_otp_secret_keys.main(['-q', '-c', '-', 'example_export.txt'])
+
+    # Assert
+    assert not file_exits('test_example_output.csv')
+
+    captured = capsys.readouterr()
+
+    expected_csv = read_csv('example_output.csv')
+    actual_csv = read_csv_str(captured.out)
+
+    assert actual_csv == expected_csv
+    assert captured.err == ''
+
+    # Clean up
+    cleanup()
+
+
 def test_keepass_csv(capsys):
     '''Two csv files .totp and .htop are generated.'''
     # Arrange
@@ -100,6 +122,31 @@ def test_keepass_csv(capsys):
     cleanup()
 
 
+def test_keepass_csv_stdout(capsys):
+    '''Two csv files .totp and .htop are generated.'''
+    # Arrange
+    cleanup()
+
+    # Act
+    extract_otp_secret_keys.main(['-q', '-k', '-', 'test/example_export_only_totp.txt'])
+
+    # Assert
+    expected_totp_csv = read_csv('example_keepass_output.totp.csv')
+    expected_hotp_csv = read_csv('example_keepass_output.hotp.csv')
+    assert not file_exits('test_example_keepass_output.totp.csv')
+    assert not file_exits('test_example_keepass_output.hotp.csv')
+    assert not file_exits('test_example_keepass_output.csv')
+
+    captured = capsys.readouterr()
+    actual_totp_csv = read_csv_str(captured.out)
+
+    assert actual_totp_csv == expected_totp_csv
+    assert captured.err == ''
+
+    # Clean up
+    cleanup()
+
+
 def test_single_keepass_csv(capsys):
     '''Does not add .totp or .hotp pre-suffix'''
     # Arrange
@@ -147,6 +194,26 @@ def test_extract_json(capsys):
     cleanup()
 
 
+def test_extract_json_stdout(capsys):
+    # Arrange
+    cleanup()
+
+    # Act
+    extract_otp_secret_keys.main(['-q', '-j', '-', 'example_export.txt'])
+
+    # Assert
+    expected_json = read_json('example_output.json')
+    assert not file_exits('test_example_output.json')
+    captured = capsys.readouterr()
+    actual_json = read_json_str(captured.out)
+
+    assert actual_json == expected_json
+    assert captured.err == ''
+
+    # Clean up
+    cleanup()
+
+
 def test_extract_not_encoded_plus(capsys):
     # Act
     extract_otp_secret_keys.main(['test/test_plus_problem_export.txt'])
@@ -265,8 +332,9 @@ def test_verbose_and_quiet(capsys):
     # Assert
     captured = capsys.readouterr()
 
-    assert len(captured.out) > 0
+    assert len(captured.err) > 0
-    assert 'The arguments --verbose and --quiet are mutually exclusive.' in captured.out
+    assert 'The arguments --verbose and --quiet are mutually exclusive.' in captured.err
+    assert captured.out == ''
 
 
 def test_wrong_data(capsys):
@@ -277,13 +345,13 @@ def test_wrong_data(capsys):
     # Assert
     captured = capsys.readouterr()
 
-    expected_stdout = '''
+    expected_stderr = '''
 ERROR: Cannot decode otpauth-migration migration payload.
 data=XXXX
 '''
 
-    assert captured.out == expected_stdout
+    assert captured.err == expected_stderr
-    assert captured.err == ''
+    assert captured.out == ''
 
 
 def test_wrong_content(capsys):
@@ -294,7 +362,7 @@ def test_wrong_content(capsys):
     # Assert
     captured = capsys.readouterr()
 
-    expected_stdout = '''
+    expected_stderr = '''
 WARN: line is not a otpauth-migration:// URL
 input file: test/test_export_wrong_content.txt
 line "Lorem ipsum dolor sit amet, consetetur sadipscing elitr, sed diam nonumy eirmod tempor invidunt ut labore et dolore magna aliquyam erat, sed diam voluptua."
@@ -306,8 +374,8 @@ line "Lorem ipsum dolor sit amet, consetetur sadipscing elitr, sed diam nonumy e
 Probably a wrong file was given
 '''
 
-    assert captured.out == expected_stdout
+    assert captured.out == ''
-    assert captured.err == ''
+    assert captured.err == expected_stderr
 
 
 def test_wrong_prefix(capsys):
@@ -317,12 +385,14 @@ def test_wrong_prefix(capsys):
     # Assert
     captured = capsys.readouterr()
 
-    expected_stdout = '''
+    expected_stderr = '''
 WARN: line is not a otpauth-migration:// URL
 input file: test/test_export_wrong_prefix.txt
 line "QR-Code:otpauth-migration://offline?data=CjUKEPqlBekzoNEukL7qlsjBCDYSDnBpQHJhc3BiZXJyeXBpGgtyYXNwYmVycnlwaSABKAEwAhABGAEgACjr4JKK%2B%2F%2F%2F%2F%2F8B"
 Probably a wrong file was given
-Name:    pi@raspberrypi
+'''
+
+    expected_stdout = '''Name:    pi@raspberrypi
 Secret:  7KSQL2JTUDIS5EF65KLMRQIIGY
 Issuer:  raspberrypi
 Type:    totp
@@ -330,7 +400,7 @@ Type:    totp
 '''
 
     assert captured.out == expected_stdout
-    assert captured.err == ''
+    assert captured.err == expected_stderr
 
 
 def test_add_pre_suffix(capsys):

utils.py

@@ -59,7 +59,7 @@ def remove_dir_with_files(dir):
 
 def read_csv(filename):
     """Returns a list of lines."""
-    with open(filename, "r") as infile:
+    with open(filename, "r", newline='') as infile:
         lines = []
         reader = csv.reader(infile)
         for line in reader:
@@ -67,12 +67,26 @@ def read_csv(filename):
         return lines
 
 
+def read_csv_str(str):
+    """Returns a list of lines."""
+    lines = []
+    reader = csv.reader(str.splitlines())
+    for line in reader:
+        lines.append(line)
+    return lines
+
+
 def read_json(filename):
     """Returns a list or a dictionary."""
     with open(filename, "r") as infile:
         return json.load(infile)
 
 
+def read_json_str(str):
+    """Returns a list or a dictionary."""
+    return json.loads(str)
+
+
 def read_file_to_list(filename):
     """Returns a list of lines."""
     with open(filename, "r") as infile: