Pkg verification interactive mode special cases

This patch adds handling for two cases for interactive
mode. 1- when non pty, the interactive mode is disabled
, and 2- timeout input function, when user fails to type
an input in interactive mode the script will continue
after a timeout.

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:

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):