diff --git a/autospec/pkg_integrity.py b/autospec/pkg_integrity.py index 95df44f..c0c1cf3 100644 --- a/autospec/pkg_integrity.py +++ b/autospec/pkg_integrity.py @@ -2,11 +2,13 @@ import os import re +import sys import argparse import shutil import tempfile import pycurl import hashlib +import signal import json from io import BytesIO from contextlib import contextmanager @@ -45,6 +47,7 @@ GNUPGCONF = """keyserver keys.gnupg.net""" PUBKEY_PATH = '/'.join([os.path.dirname(os.path.abspath(__file__)), "keyring", "{}.pkey"]) CMD_TIMEOUT = 20 ENV = os.environ +INPUT_GETTER_TIMEOUT = 60 def update_gpg_conf(proxy_value): @@ -385,23 +388,41 @@ def get_verifier(filename): return VERIFIER_TYPES.get(ext, None) +def get_input(message, default): + try: + import_key = input(message) + if import_key == '': + import_key = default + return import_key.lower() == 'y' + except: + return None + + +def input_timeout(signum, frame): + print('\ninput timed out') + raise Exception('keyboard timed out') + + class InputGetter(object): - def __init__(self, message='?', default='N'): + def __init__(self, message='?', default='N', timeout=INPUT_GETTER_TIMEOUT): self.message = message self.default = default + self.timeout = timeout + signal.signal(signal.SIGALRM, input_timeout) def get_answer(self): - import_key = input(self.message) - if import_key == '': - import_key = self.default - return import_key.lower() == 'y' + signal.alarm(self.timeout) + inpt = get_input(self.message, self.default) + signal.alarm(0) + return inpt def attempt_key_import(keyid): print(SEPT) ig = InputGetter('\nDo you want to attempt to import keyid {}: (y/N) '.format(keyid)) - if ig.get_answer() is False: + import_key_answer = ig.get_answer() + if import_key_answer in [None, False]: return False with cli_gpg_ctx() as ctx: err, _ = ctx.import_key(keyid) @@ -511,7 +532,7 @@ def from_url(url, download_path, interactive=True): verifier = get_verifier(package_name) return apply_verification(verifier, **{ 'package_path': package_path, - 'url': url, + 'url': url, 'interactive': interactive}) @@ -539,6 +560,7 @@ def check(url, download_path, interactive=True): package_name = filename_from_url(url) package_path = os.path.join(download_path, package_name) package_check = get_integrity_file(package_path) + interactive = interactive and sys.stdin.isatty() print(SEPT) print('Performing package integrity verification\n') if package_check is not None: diff --git a/tests/test_pkg_integrity.py b/tests/test_pkg_integrity.py index 26a3737..63b3ac3 100644 --- a/tests/test_pkg_integrity.py +++ b/tests/test_pkg_integrity.py @@ -110,13 +110,16 @@ class TestGPGVerifier(unittest.TestCase): pkg_integrity.attempt_to_download(XATTR_PKT_URL, out_file1) result = pkg_integrity.from_url(ALEMBIC_PKT_URL, tmpd) self.assertTrue(result) + # Monkey patching def say_no(_): return False + _ = pkg_integrity.InputGetter.get_answer pkg_integrity.InputGetter.get_answer = say_no with self.assertRaises(SystemExit) as a: pkg_integrity.from_url(XATTR_PKT_URL, tmpd) self.assertEqual(a.exception.code, 1) + pkg_integrity.InputGetter.get_answer = _ def test_from_disk(self): with tempfile.TemporaryDirectory() as tmpd: @@ -152,13 +155,26 @@ class TestGPGVerifier(unittest.TestCase): def test_pubkey_import(self): def say_yes(_): - return True + return True + _ = pkg_integrity.InputGetter.get_answer pkg_integrity.InputGetter.get_answer = say_yes keyid = '0' + KEYID[1:] result = pkg_integrity.attempt_key_import(keyid) self.assertTrue(result is False) result = pkg_integrity.attempt_key_import(KEYID) self.assertTrue(result) + pkg_integrity.InputGetter.get_answer = _ + + +class TestInputGetter(unittest.TestCase): + + def test_timput(self): + ig = pkg_integrity.InputGetter(default='N', timeout=2) + answer = ig.get_answer() + self.assertTrue(answer is None) + ig = pkg_integrity.InputGetter(default='Y', timeout=2) + answer = ig.get_answer() + self.assertTrue(answer is None) class TestUtils(unittest.TestCase):