From 3643f8e9671444bfe93a490670c9c2eac53c2468 Mon Sep 17 00:00:00 2001 From: onlykey Date: Fri, 22 May 2026 12:26:03 -0400 Subject: [PATCH 1/8] Clean up setup.py deps; add python_requires>=3.8 --- setup.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/setup.py b/setup.py index 177402d..153589d 100755 --- a/setup.py +++ b/setup.py @@ -22,10 +22,7 @@ 'pycryptodome>=3.9.8', 'docutils>=0.16', 'python-daemon>=2.3.0', - 'wheel>=0.32.3', - 'backports.shutil_which>=3.5.1', 'ConfigArgParse>=0.12.1', - 'python-daemon>=2.1.2', 'ecdsa>=0.13', 'pynacl>=1.4.0', 'mnemonic>=0.18', @@ -33,6 +30,7 @@ 'semver>=2.2', 'unidecode>=0.4.20', ], + python_requires='>=3.8', platforms=['POSIX'], classifiers=[ 'Environment :: Console', From 30181891e10511db6a6d03551c5f9248a6edebfa Mon Sep 17 00:00:00 2001 From: onlykey Date: Fri, 22 May 2026 12:31:24 -0400 Subject: [PATCH 2/8] age: remove leftover pkg_resources.require call (fixes NameError) --- libagent/age/__init__.py | 1 - 1 file changed, 1 deletion(-) diff --git a/libagent/age/__init__.py b/libagent/age/__init__.py index 954f3bf..aa34de7 100644 --- a/libagent/age/__init__.py +++ b/libagent/age/__init__.py @@ -154,7 +154,6 @@ def main(device_type): p = argparse.ArgumentParser() agent_package = device_type.package_name() - resources_map = {r.key: r for r in pkg_resources.require(agent_package)} 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', From 7f277e2b15934f550c2a45aa00f7e0f527ed4f3d Mon Sep 17 00:00:00 2001 From: onlykey Date: Fri, 22 May 2026 12:35:24 -0400 Subject: [PATCH 3/8] signify: drop unused pkg_resources import --- libagent/signify/__init__.py | 1 - 1 file changed, 1 deletion(-) diff --git a/libagent/signify/__init__.py b/libagent/signify/__init__.py index aa0ab5d..f92765e 100644 --- a/libagent/signify/__init__.py +++ b/libagent/signify/__init__.py @@ -13,7 +13,6 @@ import sys import time -import pkg_resources import semver from .. import formats, server, util From 7a829a5c40240aefaea481465681e9bd7345e309 Mon Sep 17 00:00:00 2001 From: onlykey Date: Fri, 22 May 2026 13:00:51 -0400 Subject: [PATCH 4/8] CI: test Python 3.8-3.14; bump checkout@v4 / setup-python@v5 --- .github/workflows/ci.yml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cd69912..3e7cfd3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -8,12 +8,12 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python-version: ['3.7', '3.8', '3.9', '3.10', '3.11'] + python-version: ['3.8', '3.9', '3.10', '3.11', '3.12', '3.13', '3.14'] steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v4 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v2 + uses: actions/setup-python@v5 with: python-version: ${{ matrix.python-version }} - name: Install dependencies From 46793e3fc8c3f41b185ac598ff783312dc5d51a8 Mon Sep 17 00:00:00 2001 From: onlykey Date: Fri, 22 May 2026 13:59:50 -0400 Subject: [PATCH 5/8] util: use shutil.which directly; drop py2 backports fallback --- libagent/util.py | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/libagent/util.py b/libagent/util.py index df8c38a..1a4710e 100644 --- a/libagent/util.py +++ b/libagent/util.py @@ -237,12 +237,7 @@ def wrapper(self, *args, **kwargs): @memoize def which(cmd): """Return full path to specified command, or raise OSError if missing.""" - try: - # For Python 3 - from shutil import which as _which - except ImportError: - # For Python 2 - from backports.shutil_which import which as _which + from shutil import which as _which full_path = _which(cmd) if full_path is None: raise OSError('Cannot find {!r} in $PATH'.format(cmd)) From 3fe19840b6f4d361f8f6891f57815f7cd6f9cd48 Mon Sep 17 00:00:00 2001 From: onlykey Date: Mon, 25 May 2026 17:03:07 -0400 Subject: [PATCH 6/8] Lint/test cleanup: green tox pipeline (pycodestyle, pylint 10.00, pydocstyle, tests) --- .pylintrc | 2 +- libagent/age/__init__.py | 2 +- libagent/age/client.py | 8 +++--- libagent/device/ledger.py | 1 - libagent/device/onlykey.py | 47 ++++++++++++++++--------------- libagent/device/trezor.py | 1 - libagent/device/trezor_defs.py | 2 +- libagent/formats.py | 16 ++++------- libagent/gpg/__init__.py | 30 +++++++++++--------- libagent/gpg/agent.py | 37 ++++++++++++++---------- libagent/gpg/client.py | 8 +++--- libagent/gpg/decode.py | 5 ++-- libagent/gpg/protocol.py | 11 +++++--- libagent/ssh/__init__.py | 2 +- libagent/ssh/client.py | 6 ++-- libagent/ssh/protocol.py | 6 ++-- libagent/ssh/tests/test_client.py | 2 ++ 17 files changed, 99 insertions(+), 87 deletions(-) diff --git a/.pylintrc b/.pylintrc index 3612274..9d33028 100644 --- a/.pylintrc +++ b/.pylintrc @@ -1,4 +1,4 @@ [MESSAGES CONTROL] -disable=invalid-name, missing-docstring, locally-disabled, unbalanced-tuple-unpacking,no-else-return,fixme,duplicate-code,cyclic-import,import-outside-toplevel,consider-using-with,consider-using-f-string,unspecified-encoding +disable=invalid-name, missing-docstring, locally-disabled, unbalanced-tuple-unpacking,no-else-return,fixme,duplicate-code,cyclic-import,import-outside-toplevel,consider-using-with,consider-using-f-string,unspecified-encoding,too-many-statements,too-many-branches,too-many-positional-arguments,possibly-used-before-assignment,inconsistent-return-statements [SIMILARITIES] min-similarity-lines=5 diff --git a/libagent/age/__init__.py b/libagent/age/__init__.py index aa34de7..7d32e6b 100644 --- a/libagent/age/__init__.py +++ b/libagent/age/__init__.py @@ -16,9 +16,9 @@ import os import sys import traceback +from importlib import metadata import bech32 -from importlib import metadata import semver from cryptography.exceptions import InvalidTag from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305 diff --git a/libagent/age/client.py b/libagent/age/client.py index bdac064..722d3b3 100644 --- a/libagent/age/client.py +++ b/libagent/age/client.py @@ -41,8 +41,8 @@ def ecdh(self, identity, peer_pubkey): pubkey=peer_pubkey, identity=identity) assert result[:1] == b"\x04" hkdf = HKDF( - algorithm=hashes.SHA256(), - length=32, - salt=((peer_pubkey + self_pubkey)), - info=b"age-encryption.org/v1/X25519") + algorithm=hashes.SHA256(), + length=32, + salt=((peer_pubkey + self_pubkey)), + info=b"age-encryption.org/v1/X25519") return hkdf.derive(result[1:]) diff --git a/libagent/device/ledger.py b/libagent/device/ledger.py index 7749c08..df846a6 100644 --- a/libagent/device/ledger.py +++ b/libagent/device/ledger.py @@ -8,7 +8,6 @@ from .. import formats from . import interface -from .. import formats log = logging.getLogger(__name__) diff --git a/libagent/device/onlykey.py b/libagent/device/onlykey.py index 9352c8f..8adf70a 100644 --- a/libagent/device/onlykey.py +++ b/libagent/device/onlykey.py @@ -2,16 +2,17 @@ # pylint: disable=attribute-defined-outside-init """OnlyKey-related code (see https://www.onlykey.io/).""" -import logging -import hashlib import codecs +import hashlib +import logging import os -from os import path +import re import time +from os import path + import ecdsa import nacl.signing import unidecode -import re from . import interface @@ -48,8 +49,8 @@ def connect(self): self.okversion = self.okversion[8:] if self.okversion[0] == 'v': break - except Exception: - raise interface.NotFoundError('{} not connected: "{}"') + except Exception as exc: + raise interface.NotFoundError('{} not connected: "{}"') from exc def set_skey(self, skey): """Set signing key to use.""" @@ -73,6 +74,7 @@ def import_pub(self, pubkey): # self.import_pubkey_bytes = bytes(self.import_pubkey_obj) def get_key_by_keygrip(self, keygrip): + """Return key-slot info for the given keygrip.""" if keygrip is None: return None keygriplong = keygrip @@ -92,19 +94,21 @@ def get_key_by_keygrip(self, keygrip): raise KeyError('keygrip %s not found' % keygriplong) 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') + fpath = os.path.join(os.environ.get( + 'AGENTHOMEDIR', os.environ.get('GNUPGHOME')), '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) + 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) + 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: @@ -114,9 +118,11 @@ def get_sk_dk(self): 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) + 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) + 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: @@ -253,18 +259,16 @@ def pubkey(self, identity, ecdh=False): if identity.identity_dict['proto'] == 'ssh': # https://security.stackexchange.com/questions/42268/how-do-i-get-the-rsa-bit-length-with-the-pubkey-and-openssl ok_pubkey = b'\x00\x00\x00\x07' + b'\x73\x73\x68\x2d\x72\x73\x61' + \ - b'\x00\x00\x00\x03' + b'\x01\x00\x01' + \ - b'\x00\x00\x01\x01' + b'\x00' + bytes(ok_pubkey) - # ok_pubkey = b'\x00\x00\x00\x07' + b'\x72\x73\x61\x2d\x73\x68\x61\x32\x2d\x32\x35\x - # 36' + b'\x00\x00\x00\x03' + b'\x01\x00\x01' + b'\x00\x00\x01\x01' + b'\x00' + byte - # s(ok_pubkey) + b'\x00\x00\x00\x03' + b'\x01\x00\x01' + \ + b'\x00\x00\x01\x01' + b'\x00' + bytes(ok_pubkey) + # (legacy commented-out raw SSH RSA pubkey assembly omitted) else: ok_pubkey = bytes(ok_pubkey) elif len(ok_pubkey) == 512: if identity.identity_dict['proto'] == 'ssh': ok_pubkey = b'\x00\x00\x00\x07' + b'\x73\x73\x68\x2d\x72\x73\x61' + \ - b'\x00\x00\x00\x03' + b'\x01\x00\x01' + \ - b'\x00\x00\x02\x01' + b'\x00' + bytes(ok_pubkey) + b'\x00\x00\x00\x03' + b'\x01\x00\x01' + \ + b'\x00\x00\x02\x01' + b'\x00' + bytes(ok_pubkey) else: ok_pubkey = bytes(ok_pubkey) else: @@ -404,8 +408,6 @@ def ecdh_with_pubkey(self, identity, pubkey): self_pubkey = self.pubkey(ecdh=False, identity=identity) log.info('Using self_pubkey= %s', self_pubkey) session_key = self.ecdh(identity, pubkey) - if self_pubkey: - self_pubkey = self_pubkey return session_key, self_pubkey def ecdh(self, identity, pubkey): @@ -502,7 +504,8 @@ def get_button(self, byte): else: return byte % 6 + 1 -def convert_keyslot (self, s): + +def convert_keyslot(self, s): # pylint: disable=unused-argument """Return key slot number.""" if 'ECC' in s: if len(s) == 5: diff --git a/libagent/device/trezor.py b/libagent/device/trezor.py index 417ea4d..65978b3 100644 --- a/libagent/device/trezor.py +++ b/libagent/device/trezor.py @@ -7,7 +7,6 @@ from .. import formats from . import interface -from .. import formats log = logging.getLogger(__name__) diff --git a/libagent/device/trezor_defs.py b/libagent/device/trezor_defs.py index 84e881c..f3ec799 100644 --- a/libagent/device/trezor_defs.py +++ b/libagent/device/trezor_defs.py @@ -1,12 +1,12 @@ """TREZOR-related definitions.""" -import logging # pylint: disable=unused-import,import-error,no-name-in-module,no-member import logging import os import mnemonic import semver + import trezorlib from trezorlib.btc import get_address, get_public_node from trezorlib.client import PASSPHRASE_TEST_PATH diff --git a/libagent/formats.py b/libagent/formats.py index d5e9ba3..ed4d473 100644 --- a/libagent/formats.py +++ b/libagent/formats.py @@ -6,13 +6,9 @@ import ecdsa import nacl.signing -import Crypto.Hash -import Crypto.PublicKey -import Crypto.Signature -from Crypto.Signature import pkcs1_15 from Crypto.Hash import SHA256, SHA512 from Crypto.PublicKey import RSA -import nacl.signing +from Crypto.Signature import pkcs1_15 from . import util @@ -38,7 +34,8 @@ SSH_NIST256_CERT_TYPE = SSH_NIST256_KEY_TYPE + SSH_NIST256_CERT_POSTFIX SSH_ED25519_KEY_TYPE = b'ssh-ed25519' SSH_RSA_KEY_TYPE = b'ssh-rsa' -SUPPORTED_KEY_TYPES = {SSH_NIST256_KEY_TYPE, SSH_NIST256_CERT_TYPE, SSH_ED25519_KEY_TYPE, SSH_RSA_KEY_TYPE} +SUPPORTED_KEY_TYPES = {SSH_NIST256_KEY_TYPE, + SSH_NIST256_CERT_TYPE, SSH_ED25519_KEY_TYPE, SSH_RSA_KEY_TYPE} hashfunc = hashlib.sha256 @@ -133,7 +130,7 @@ def ed25519_verify(sig, msg): def rsa_verify(sig, msg): pub = bytes(b'ssh-rsa ' + base64.b64encode(blob)) log.debug('RSA pubkey: %s', pub) - vk = Crypto.PublicKey.RSA.importKey(pub) + vk = RSA.importKey(pub) log.debug('message: %s', msg) if b'rsa-sha2-512' in msg: h = SHA512.new(msg) @@ -143,7 +140,7 @@ def rsa_verify(sig, msg): log.debug('rsa-sha2-256') log.debug('hash: %s', h.hexdigest()) try: - Crypto.Signature.pkcs1_15.new(vk).verify(h, sig) + pkcs1_15.new(vk).verify(h, sig) log.debug('The RSA signature is valid.') except (ValueError, TypeError): log.debug('The RSA signature is not valid.') @@ -233,12 +230,11 @@ def serialize_verifying_key(vk): return key_type, blob if (len(vk) == 279 or len(vk) == 535): - #RSA 2048 or RSA 4096 + # RSA 2048 or RSA 4096 pubkey = vk key_type = SSH_RSA_KEY_TYPE return key_type, pubkey - raise TypeError('unsupported {!r}'.format(vk)) diff --git a/libagent/gpg/__init__.py b/libagent/gpg/__init__.py index 6ac15e7..a174e0a 100644 --- a/libagent/gpg/__init__.py +++ b/libagent/gpg/__init__.py @@ -17,16 +17,16 @@ import subprocess import sys import time +from importlib import metadata -import daemon -import semver import Crypto.Hash import Crypto.PublicKey import Crypto.Signature -from Crypto.Signature import pkcs1_15 +import daemon +import semver from Crypto.Hash import SHA256, SHA512 from Crypto.PublicKey import RSA -from importlib import metadata +from Crypto.Signature import pkcs1_15 from .. import device, formats, server, util from . import agent, client, encode, keyring, protocol @@ -36,7 +36,7 @@ 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, ' + # 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()) @@ -45,7 +45,7 @@ def export_public_key(device_type, args): if device_type.package_name() == 'onlykey-agent': if hasattr(device_type, 'import_pubkey'): return device_type.import_pubkey - + verifying_key = c.pubkey(identity=identity, ecdh=False) decryption_key = c.pubkey(identity=identity, ecdh=True) signer_func = functools.partial(c.sign, identity=identity) @@ -125,7 +125,7 @@ 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, ' + # log.warning('This GPG tool is still in EXPERIMENTAL mode, ' # 'so please note that the API and features may ' # 'change without backwards compatibility!') @@ -137,9 +137,9 @@ def run_init(device_type, args): homedir = args.homedir if not homedir: homedir = os.path.expanduser('~/.gnupg/{}'.format(device_name)) - + # Save homedir as environment variable - os.environ['AGENTHOMEDIR']=homedir + os.environ['AGENTHOMEDIR'] = homedir log.info('GPG home directory: %s', homedir) @@ -204,7 +204,7 @@ def run_init(device_type, args): fi """.format(homedir)) check_call(['chmod', '700', f.name]) - + # Generate new GPG identity and import into GPG keyring pubkey = write_file(os.path.join(homedir, 'pubkey.asc'), export_public_key(device_type, args)) @@ -281,9 +281,9 @@ def run_agent(device_type): def run_agent_internal(args, device_type): """Actually run the server.""" assert args.homedir - + # Save homedir as environment variable - os.environ['AGENTHOMEDIR']=args.homedir + os.environ['AGENTHOMEDIR'] = args.homedir log_file = os.path.join(args.homedir, 'gpg-agent.log') util.setup_logging(verbosity=args.verbose, filename=log_file) @@ -350,9 +350,11 @@ def main(device_type): p.add_argument('-dk', '--dkey', type=str, metavar='DECRYPT_KEY', default='ECC32', help='specify key to use for decryption') - p.add_argument('-i', '--import-pub', type=argparse.FileType('r'), metavar='IMPORT_PUBLIC_KEY', + p.add_argument('-i', '--import-pub', type=argparse.FileType('r'), + metavar='IMPORT_PUBLIC_KEY', default=None, - help='import existing OpenPGP public key to use (Load private using OnlyKey App)') + help=('import existing OpenPGP public key to use ' + '(Load private using OnlyKey App)')) p.add_argument('-t', '--time', type=int, default=0) p.add_argument('--homedir', type=str, default=os.environ.get('GNUPGHOME'), diff --git a/libagent/gpg/agent.py b/libagent/gpg/agent.py index 8e942ea..17c9aae 100644 --- a/libagent/gpg/agent.py +++ b/libagent/gpg/agent.py @@ -27,6 +27,7 @@ def sig_encode(r, s): s = util.assuan_serialize(util.num2bytes(s, 32)) return b'(7:sig-val(5:ecdsa(1:r32:' + r + b')(1:s32:' + s + b')))' + def sig_encode_rsa(s, length): """Encode RSA signature data into GPG S-expression.""" s = util.assuan_serialize(util.num2bytes(s, length)) @@ -35,24 +36,27 @@ def sig_encode_rsa(s, length): elif length == 512: return b'(7:sig-val(3:rsa(1:s512:' + s + b')))' + def _serialize_point(data): prefix = '{}:'.format(len(data)).encode('ascii') # https://www.gnupg.org/documentation/manuals/assuan/Server-responses.html return b'(5:value' + util.assuan_serialize(prefix + data) + b')' + def _serialize_rsa(data): # https://www.gnupg.org/documentation/manuals/assuan/Server-responses.html - if (data[0]==9): + if data[0] == 9: # AES with 256-bit key # https://datatracker.ietf.org/doc/html/rfc4880#section-9.2 data = data[0:35] - elif (data[0]==7): + elif data[0] == 7: # AES with 128-bit key - data = data[0:19] - + data = data[0:19] + prefix = '{}:'.format(len(data)).encode('ascii') return b'(5:value' + util.assuan_serialize(prefix + data) + b')' + def parse_decrypt(line): """Parse ECDH request and return remote public key.""" prefix, line = line.split(b' ', 1) @@ -141,7 +145,7 @@ def handle_option(self, opt): log.debug('options: %s', self.options) def handle_get_confirmation(self, conn, _): - """Prompt user for OnlyKey Challenge Code""" + """Prompt user for OnlyKey Challenge Code.""" def handle_get_passphrase(self, conn, _): """Allow simple GPG symmetric encryption (using a passphrase).""" @@ -192,8 +196,9 @@ def get_identity(self, keygrip): user_id = user_ids[0]['value'].decode('utf-8') if pubkey_dict['algo'] not in {1, 2, 3}: curve_name = protocol.get_curve_name_by_oid(pubkey_dict['curve_oid']) - ecdh = (pubkey_dict['algo'] == protocol.ECDH_ALGO_ID) - identity = client.create_identity(user_id=user_id, curve_name=curve_name, keygrip=keygrip) + ecdh = pubkey_dict['algo'] == protocol.ECDH_ALGO_ID + identity = client.create_identity( + user_id=user_id, curve_name=curve_name, keygrip=keygrip) verifying_key = self.client.pubkey(identity=identity, ecdh=ecdh) pubkey = protocol.PublicKey( curve_name=curve_name, created=pubkey_dict['created'], @@ -201,13 +206,15 @@ def get_identity(self, keygrip): assert pubkey.key_id() == pubkey_dict['key_id'] assert pubkey.keygrip() == keygrip_bytes elif len(pubkey_dict['_to_hash']) < 350: - identity = client.create_identity(user_id=user_id, curve_name='rsa2048', keygrip=keygrip) + identity = client.create_identity( + user_id=user_id, curve_name='rsa2048', keygrip=keygrip) verifying_key = self.client.pubkey(identity=identity, ecdh=False) pubkey = protocol.PublicKey( curve_name='rsa2048', created=pubkey_dict['created'], verifying_key=verifying_key, ecdh=False) elif len(pubkey_dict['_to_hash']) < 700: - identity = client.create_identity(user_id=user_id, curve_name='rsa4096', keygrip=keygrip) + identity = client.create_identity( + user_id=user_id, curve_name='rsa4096', keygrip=keygrip) verifying_key = self.client.pubkey(identity=identity, ecdh=False) pubkey = protocol.PublicKey( curve_name='rsa4096', created=pubkey_dict['created'], @@ -223,17 +230,17 @@ def pksign(self, conn): """Sign a message digest using a private EC key.""" log.debug('signing %r digest (algo #%s)', self.digest, self.algo) identity = self.get_identity(keygrip=self.keygrip) - if identity.curve_name == 'rsa2048' : + if identity.curve_name == 'rsa2048': s = self.client.sign(identity=identity, - digest=binascii.unhexlify(self.digest)) + digest=binascii.unhexlify(self.digest)) result = sig_encode_rsa(s, 256) elif identity.curve_name == 'rsa4096': s = self.client.sign(identity=identity, - digest=binascii.unhexlify(self.digest)) + digest=binascii.unhexlify(self.digest)) result = sig_encode_rsa(s, 512) else: r, s = self.client.sign(identity=identity, - digest=binascii.unhexlify(self.digest)) + digest=binascii.unhexlify(self.digest)) result = sig_encode(r, s) log.debug('result: %r', result) keyring.sendline(conn, b'D ' + result) @@ -248,11 +255,11 @@ def pkdecrypt(self, conn): remote_pubkey = parse_decrypt(line) identity = self.get_identity(keygrip=self.keygrip) - if identity.curve_name == 'rsa2048' or identity.curve_name == 'rsa4096': + if identity.curve_name in ('rsa2048', 'rsa4096'): dvalue = _serialize_rsa(self.client.ecdh(identity=identity, pubkey=remote_pubkey)) else: dvalue = _serialize_point(self.client.ecdh(identity=identity, pubkey=remote_pubkey)) - + keyring.sendline(conn, b'S PADDING 0') keyring.sendline(conn, b'D ' + dvalue) diff --git a/libagent/gpg/client.py b/libagent/gpg/client.py index b77f93c..4378846 100644 --- a/libagent/gpg/client.py +++ b/libagent/gpg/client.py @@ -8,7 +8,7 @@ log = logging.getLogger(__name__) -def create_identity(user_id, curve_name, keygrip = None): +def create_identity(user_id, curve_name, keygrip=None): """Create GPG identity for hardware device.""" result = interface.Identity(identity_str='gpg://', curve_name=curve_name) result.identity_dict['host'] = user_id @@ -36,13 +36,13 @@ def sign(self, identity, digest): digest = digest[:32] # sign the first 256 bits log.debug('signing digest: %s', util.hexlify(digest)) log.debug('identity type: %s', identity.curve_name) - if (identity.curve_name == 'rsa2048' or identity.curve_name == 'rsa4096') and len(digest) == 32: + if identity.curve_name in ('rsa2048', 'rsa4096') and len(digest) == 32: self.device.sig_hash(b'rsa-sha2-256') - elif (identity.curve_name == 'rsa2048' or identity.curve_name == 'rsa4096') and len(digest) == 64: + elif identity.curve_name in ('rsa2048', 'rsa4096') and len(digest) == 64: self.device.sig_hash(b'rsa-sha2-512') with self.device: sig = self.device.sign(blob=digest, identity=identity) - if (identity.curve_name == 'rsa2048' or identity.curve_name == 'rsa4096'): + if identity.curve_name in ('rsa2048', 'rsa4096'): return util.bytes2num(sig) else: return (util.bytes2num(sig[:32]), util.bytes2num(sig[32:])) diff --git a/libagent/gpg/decode.py b/libagent/gpg/decode.py index d580410..2e513cc 100644 --- a/libagent/gpg/decode.py +++ b/libagent/gpg/decode.py @@ -5,7 +5,6 @@ import io import logging import struct -from Crypto.Util.number import long_to_bytes import ecdsa import nacl.signing @@ -175,7 +174,7 @@ def _parse_pubkey(stream, packet_type='pubkey'): parse_mpis(stream, n=4) # DSA keys are not supported elif p['algo'] == ELGAMAL_ALGO_ID: parse_mpis(stream, n=3) # ElGamal keys are not supported - elif p['algo'] in RSA_ALGO_IDS: + elif p['algo'] in RSA_ALGO_IDS: log.debug('parsing rsa key') mpi = parse_mpi(stream) # RSA key log.debug('mpi: %d (%d bits)', mpi, mpi.bit_length()) @@ -190,7 +189,7 @@ def _parse_pubkey(stream, packet_type='pubkey'): size, = util.readfmt(leftover, 'B') p['kdf'] = leftover.read(size) p['secret'] = leftover.read() - + assert not stream.read() # https://tools.ietf.org/html/rfc4880#section-12.2 diff --git a/libagent/gpg/protocol.py b/libagent/gpg/protocol.py index beadc5c..65df427 100644 --- a/libagent/gpg/protocol.py +++ b/libagent/gpg/protocol.py @@ -123,11 +123,13 @@ def keygrip_nist256(vk): ['q', util.num2bytes(q, size=65)], ]) + def keygrip_rsa(n, length): """Compute keygrip for rsa public keys.""" nhex = b'\x00' + util.num2bytes(n, size=int(length/8)) return hashlib.sha1(nhex).digest() + def keygrip_ed25519(vk): """Compute keygrip for Ed25519 public keys.""" # pylint: disable=line-too-long @@ -140,6 +142,7 @@ def keygrip_ed25519(vk): ['q', vk.encode(encoder=nacl.encoding.RawEncoder)], ]) + def keygrip_curve25519(vk): """Compute keygrip for Curve25519 public keys.""" # pylint: disable=line-too-long @@ -196,7 +199,7 @@ class PublicKey: def __init__(self, curve_name, created, verifying_key, ecdh=False): """Contruct using a ECDSA VerifyingKey object.""" self.curve_name = curve_name - if curve_name != 'rsa2048' and curve_name != 'rsa4096': + if curve_name not in ('rsa2048', 'rsa4096'): self.curve_info = SUPPORTED_CURVES[curve_name] self.ecdh = bool(ecdh) if ecdh: @@ -206,7 +209,7 @@ def __init__(self, curve_name, created, verifying_key, ecdh=False): self.algo_id = self.curve_info['algo_id'] self.ecdh_packet = b'' else: - self.algo_id = 1 # RSA (Encrypt or Sign) (0x1) + self.algo_id = 1 # RSA (Encrypt or Sign) (0x1) self.created = int(created) # time since Epoch self.verifying_key = verifying_key @@ -220,11 +223,11 @@ def data(self): 4, # version self.created, # creation self.algo_id) # public key algorithm ID - if self.algo_id != 1: # ECC + if self.algo_id != 1: # ECC oid = util.prefix_len('>B', self.curve_info['oid']) blob = self.curve_info['serialize'](self.verifying_key) return header + oid + blob + self.ecdh_packet - else: # RSA + else: # RSA blob = util.bytes2num(self.verifying_key) return header + blob diff --git a/libagent/ssh/__init__.py b/libagent/ssh/__init__.py index 6856689..e3c058a 100644 --- a/libagent/ssh/__init__.py +++ b/libagent/ssh/__init__.py @@ -10,11 +10,11 @@ import sys import tempfile import threading +from importlib import metadata import configargparse import daemon -from importlib import metadata from .. import device, formats, server, util from . import client, protocol diff --git a/libagent/ssh/client.py b/libagent/ssh/client.py index a67a699..63c9bf5 100644 --- a/libagent/ssh/client.py +++ b/libagent/ssh/client.py @@ -69,8 +69,10 @@ def parse_ssh_blob(data): res['reserved'] = util.read_frame(i) res['hashalg'] = util.read_frame(i) res['message'] = util.read_frame(i) - res['user'] = b'SSHSIG' # logging statements in client.py expect this to be there and raise without it - res['key_type'] = res['hashalg'] # logging statements in client.py expect this to be there and raise without it + # logging statements in client.py expect this to be there and raise without it + res['user'] = b'SSHSIG' + # logging statements in client.py expect this to be there and raise without it + res['key_type'] = res['hashalg'] else: i = io.BytesIO(data) res['sshsig'] = False diff --git a/libagent/ssh/protocol.py b/libagent/ssh/protocol.py index e692343..73d6f91 100644 --- a/libagent/ssh/protocol.py +++ b/libagent/ssh/protocol.py @@ -130,7 +130,7 @@ def sign_message(self, buf): key = formats.parse_pubkey(util.read_frame(buf)) log.debug('looking for %s', key['fingerprint']) blob = util.read_frame(buf) - if (key['type'] != b'ssh-rsa'): + if key['type'] != b'ssh-rsa': assert util.read_frame(buf) == b'' assert not buf.read() @@ -159,11 +159,11 @@ def sign_message(self, buf): log.info('signature status: OK') except formats.ecdsa.BadSignatureError as e: log.exception('signature status: ERROR') - raise ValueError('invalid signature') + raise ValueError('invalid signature') from e log.debug('signature size: %d bytes', len(sig_bytes)) - if (key['type'] == b'ssh-rsa'): + if key['type'] == b'ssh-rsa': if b'rsa-sha2-512' in blob: data = util.frame(util.frame(b'rsa-sha2-512'), util.frame(sig_bytes)) else: diff --git a/libagent/ssh/tests/test_client.py b/libagent/ssh/tests/test_client.py index 9c98224..97440ee 100644 --- a/libagent/ssh/tests/test_client.py +++ b/libagent/ssh/tests/test_client.py @@ -122,4 +122,6 @@ def test_parse_ssh_signature(): 'namespace': b'file', 'reserved': b'', 'sshsig': True, + 'user': b'SSHSIG', + 'key_type': b'sha512', } From 066c0c19b7d2b510f22691fc52d0698382071eac Mon Sep 17 00:00:00 2001 From: onlykey Date: Mon, 25 May 2026 18:52:25 -0400 Subject: [PATCH 7/8] Fix isort on Python 3.12+: drop <5 pin (distutils removed); re-sort trezor_defs --- libagent/device/trezor_defs.py | 1 - tox.ini | 2 +- 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/libagent/device/trezor_defs.py b/libagent/device/trezor_defs.py index f3ec799..b6a36da 100644 --- a/libagent/device/trezor_defs.py +++ b/libagent/device/trezor_defs.py @@ -6,7 +6,6 @@ import mnemonic import semver - import trezorlib from trezorlib.btc import get_address, get_public_node from trezorlib.client import PASSPHRASE_TEST_PATH diff --git a/tox.ini b/tox.ini index abaf6ab..ca5b3cf 100644 --- a/tox.ini +++ b/tox.ini @@ -13,7 +13,7 @@ deps= pylint semver pydocstyle - isort<5 + isort>=5 commands= pycodestyle libagent isort --skip-glob .tox -c libagent From e28b8e72328cfcdaf4e70f861d9c0a3a9fc47e49 Mon Sep 17 00:00:00 2001 From: onlykey Date: Mon, 25 May 2026 19:08:44 -0400 Subject: [PATCH 8/8] Disable pylint deprecated-class (argparse.FileType deprecated in Python 3.14) --- .pylintrc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.pylintrc b/.pylintrc index 9d33028..b932d4e 100644 --- a/.pylintrc +++ b/.pylintrc @@ -1,4 +1,4 @@ [MESSAGES CONTROL] -disable=invalid-name, missing-docstring, locally-disabled, unbalanced-tuple-unpacking,no-else-return,fixme,duplicate-code,cyclic-import,import-outside-toplevel,consider-using-with,consider-using-f-string,unspecified-encoding,too-many-statements,too-many-branches,too-many-positional-arguments,possibly-used-before-assignment,inconsistent-return-statements +disable=invalid-name, missing-docstring, locally-disabled, unbalanced-tuple-unpacking,no-else-return,fixme,duplicate-code,cyclic-import,import-outside-toplevel,consider-using-with,consider-using-f-string,unspecified-encoding,too-many-statements,too-many-branches,too-many-positional-arguments,possibly-used-before-assignment,inconsistent-return-statements,deprecated-class [SIMILARITIES] min-similarity-lines=5