diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3e7cfd3..61cfdc7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -11,9 +11,9 @@ jobs: python-version: ['3.8', '3.9', '3.10', '3.11', '3.12', '3.13', '3.14'] steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v5 + uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0 with: python-version: ${{ matrix.python-version }} - name: Install dependencies diff --git a/libagent/age/__init__.py b/libagent/age/__init__.py index 7d32e6b..3baea2d 100644 --- a/libagent/age/__init__.py +++ b/libagent/age/__init__.py @@ -16,7 +16,6 @@ import os import sys import traceback -from importlib import metadata import bech32 import semver @@ -154,10 +153,8 @@ def main(device_type): p = argparse.ArgumentParser() agent_package = device_type.package_name() - resources = [metadata.distribution(agent_package), metadata.distribution('lib-agent')] - versions = '\n'.join('{}={}'.format(r.metadata['Name'], r.version) for r in resources) p.add_argument('--version', help='print the version info', - action='version', version=versions) + action='version', version=util.format_versions(agent_package)) p.add_argument('-i', '--identity') p.add_argument('-v', '--verbose', default=0, action='count') diff --git a/libagent/device/onlykey.py b/libagent/device/onlykey.py index 8adf70a..6345142 100644 --- a/libagent/device/onlykey.py +++ b/libagent/device/onlykey.py @@ -50,7 +50,8 @@ def connect(self): if self.okversion[0] == 'v': break except Exception as exc: - raise interface.NotFoundError('{} not connected: "{}"') from exc + raise interface.NotFoundError( + '{} not connected: "{}"'.format(self.device_name, exc)) from exc def set_skey(self, skey): """Set signing key to use.""" @@ -94,45 +95,58 @@ def get_key_by_keygrip(self, keygrip): raise KeyError('keygrip %s not found' % keygriplong) return None + DEFAULT_SLOT = 132 + + _SKEY_SLOT_RE = re.compile(r'--skey-slot=(\S+)') + _DKEY_SLOT_RE = re.compile(r'--dkey-slot=(\S+)') + + @staticmethod + def _parse_slot_value(value): + """Convert a --(s|d)key-slot=VALUE token (e.g. 'ECC1', 'RSA2', '132'). + + Returns an int slot number, or None if VALUE can't be parsed. + """ + if not value: + return None + try: + if value.startswith('ECC'): + return int(value[3:]) + 100 + if value.startswith('RSA'): + return int(value[3:]) + return int(value) + except ValueError: + log.warning('Unrecognized key-slot value %r in run-agent.sh', value) + return None + def get_sk_dk(self): """Get signing key and decryption key slots from config.""" - fpath = os.path.join(os.environ.get( - 'AGENTHOMEDIR', os.environ.get('GNUPGHOME')), 'run-agent.sh') + homedir = os.environ.get('AGENTHOMEDIR') or os.environ.get('GNUPGHOME') + if not homedir: + log.debug( + 'Neither AGENTHOMEDIR nor GNUPGHOME set; using default slot %d', + self.DEFAULT_SLOT) + self.set_skey(self.DEFAULT_SLOT) + self.set_dkey(self.DEFAULT_SLOT) + return + + fpath = os.path.join(homedir, 'run-agent.sh') log.debug('Path to run-agent.sh = %s', fpath) - if path.exists(fpath): - with open(fpath) as f: - s = f.read() - if '--skey-slot=ECC' in s: - if s[s.find('--skey-slot=')+16:s.find('--skey-slot=')+17] == ' ': - self.set_skey( - int(s[s.find('--skey-slot=')+15:s.find('--skey-slot=')+16])+100) - else: - self.set_skey( - int(s[s.find('--skey-slot=')+15:s.find('--skey-slot=')+17])+100) - elif '--skey-slot=RSA' in s: - self.set_skey(int(s[s.find('--skey-slot=')+15:s.find('--skey-slot=')+16])) - elif '--skey-slot=' in s: - if s[s.find('--skey-slot=')+13:s.find('--skey-slot=')+14] == ' ': - self.set_skey(int(s[s.find('--skey-slot=')+12:s.find('--skey-slot=')+13])) - else: - self.set_skey(int(s[s.find('--skey-slot=')+12:s.find('--skey-slot=')+15])) - if '--dkey-slot=ECC' in s: - if s[s.find('--dkey-slot=')+16:s.find('--dkey-slot=')+17] == ' ': - self.set_dkey( - int(s[s.find('--dkey-slot=')+15:s.find('--dkey-slot=')+16])+100) - else: - self.set_dkey( - int(s[s.find('--dkey-slot=')+15:s.find('--dkey-slot=')+17])+100) - elif '--dkey-slot=RSA' in s: - self.set_dkey(int(s[s.find('--dkey-slot=')+15:s.find('--dkey-slot=')+16])) - elif '--dkey-slot=' in s: - if s[s.find('--dkey-slot=')+13:s.find('--dkey-slot=')+14] == ' ': - self.set_dkey(int(s[s.find('--dkey-slot=')+12:s.find('--dkey-slot=')+13])) - else: - self.set_dkey(int(s[s.find('--dkey-slot=')+12:s.find('--dkey-slot=')+15])) - else: - self.set_skey(132) - self.set_dkey(132) + if not path.exists(fpath): + self.set_skey(self.DEFAULT_SLOT) + self.set_dkey(self.DEFAULT_SLOT) + return + + with open(fpath) as f: + content = f.read() + + skey_match = self._SKEY_SLOT_RE.search(content) + dkey_match = self._DKEY_SLOT_RE.search(content) + + skey = self._parse_slot_value(skey_match.group(1)) if skey_match else None + dkey = self._parse_slot_value(dkey_match.group(1)) if dkey_match else None + + self.set_skey(skey if skey is not None else self.DEFAULT_SLOT) + self.set_dkey(dkey if dkey is not None else self.DEFAULT_SLOT) def sig_hash(self, sighash): """Set signature hashing algorithm to use.""" diff --git a/libagent/gpg/__init__.py b/libagent/gpg/__init__.py index a174e0a..53ab255 100644 --- a/libagent/gpg/__init__.py +++ b/libagent/gpg/__init__.py @@ -17,7 +17,6 @@ import subprocess import sys import time -from importlib import metadata import Crypto.Hash import Crypto.PublicKey @@ -36,9 +35,6 @@ def export_public_key(device_type, args): """Generate a new pubkey for a new/existing GPG identity.""" - # log.warning('NOTE: in order to re-generate the exact same GPG key later, ' - # 'run this command with "--time=%d" commandline flag (to set ' - # 'the timestamp of the GPG key manually).', args.time) c = client.Client(device=device_type()) identity = client.create_identity(user_id=args.user_id, curve_name=args.ecdsa_curve) @@ -125,9 +121,6 @@ def write_file(path, data): def run_init(device_type, args): """Initialize hardware-based GnuPG identity.""" util.setup_logging(verbosity=args.verbose) - # log.warning('This GPG tool is still in EXPERIMENTAL mode, ' - # 'so please note that the API and features may ' - # 'change without backwards compatibility!') verify_gpg_version() @@ -310,7 +303,6 @@ def run_agent_internal(args, device_type): handler.handle(conn) except agent.AgentStop: log.info('stopping gpg-agent') - return except IOError as e: log.info('connection closed: %s', e) return @@ -328,10 +320,8 @@ def main(device_type): parser = argparse.ArgumentParser(epilog=epilog) agent_package = device_type.package_name() - resources = [metadata.distribution(agent_package), metadata.distribution('lib-agent')] - versions = '\n'.join('{}={}'.format(r.metadata['Name'], r.version) for r in resources) parser.add_argument('--version', help='print the version info', - action='version', version=versions) + action='version', version=util.format_versions(agent_package)) subparsers = parser.add_subparsers(title='Action', dest='action') subparsers.required = True diff --git a/libagent/ssh/__init__.py b/libagent/ssh/__init__.py index e3c058a..2699b06 100644 --- a/libagent/ssh/__init__.py +++ b/libagent/ssh/__init__.py @@ -10,7 +10,6 @@ import sys import tempfile import threading -from importlib import metadata import configargparse import daemon @@ -72,10 +71,8 @@ def create_agent_parser(device_type): p.add_argument('-v', '--verbose', default=0, action='count') agent_package = device_type.package_name() - resources = [metadata.distribution(agent_package), metadata.distribution('lib-agent')] - versions = '\n'.join('{}={}'.format(r.metadata['Name'], r.version) for r in resources) p.add_argument('--version', help='print the version info', - action='version', version=versions) + action='version', version=util.format_versions(agent_package)) curve_names = ', '.join(sorted(formats.SUPPORTED_CURVES)) diff --git a/libagent/util.py b/libagent/util.py index 1a4710e..4e6c14b 100644 --- a/libagent/util.py +++ b/libagent/util.py @@ -245,6 +245,23 @@ def which(cmd): return full_path +def format_versions(agent_package): + """Return a newline-separated 'name=version' string for the agent + lib-agent. + + Falls back to '(version unknown)' if a distribution isn't installed + (e.g. running from a source checkout without ``pip install -e .``). + """ + from importlib import metadata + lines = [] + for name in (agent_package, 'lib-agent'): + try: + dist = metadata.distribution(name) + lines.append('{}={}'.format(dist.metadata['Name'], dist.version)) + except metadata.PackageNotFoundError: + lines.append('{}=(version unknown)'.format(name)) + return '\n'.join(lines) + + def assuan_serialize(data): """Serialize data according to ASSUAN protocol (for GPG daemon communication).""" for c in [b'%', b'\n', b'\r']: diff --git a/setup.py b/setup.py index 153589d..a542eda 100755 --- a/setup.py +++ b/setup.py @@ -3,7 +3,7 @@ setup( name='lib-agent', - version='1.0.7', + version='1.0.8', description='Using OnlyKey as hardware SSH and GPG agent', author='CryptoTrust', author_email='admin@crp.to', diff --git a/tox.ini b/tox.ini index ca5b3cf..150329f 100644 --- a/tox.ini +++ b/tox.ini @@ -13,7 +13,7 @@ deps= pylint semver pydocstyle - isort>=5 + isort>=5,<7 commands= pycodestyle libagent isort --skip-glob .tox -c libagent