Compare commits

...

176 Commits

Author SHA1 Message Date
Roman Zeyde
71a8930021 bump version 2017-01-05 23:12:54 +02:00
Roman Zeyde
74e8f21a22 gpg: export secret subkey 2017-01-01 18:14:52 +02:00
Roman Zeyde
897236d556 gpg: allow decoding secret keys 2017-01-01 18:14:28 +02:00
Roman Zeyde
5bec0e8382 README: upgrade also pip 2017-01-01 10:08:19 +02:00
Roman Zeyde
3cb7f6fd21 gpg: export secret primary key 2016-12-30 18:55:18 +02:00
Roman Zeyde
cad2ec1239 device: import device-specific defs module lazily
It may fail on unsupported platforms (e.g. keepkeylib does not supoprt Python 3)
2016-12-27 12:34:07 +02:00
Roman Zeyde
604b2b7e99 gpg: allow GPG 2.1.11+ (to support Ubuntu 16.04 & Mint 18) 2016-12-27 10:12:34 +02:00
Roman Zeyde
159bd79b5f gpg: list fingerpints explicitly during init 2016-12-27 10:04:48 +02:00
Roman Zeyde
dde0b60e83 Merge pull request #87 from aceat64/bugfix/mosh
Mosh doesn't support "-l" for user, only user@host for args
2016-12-15 19:25:19 +02:00
Andrew LeCody
109bb3b47f Mosh doesn't support "-l" for user, only user@host for args 2016-12-14 23:59:45 -06:00
Roman Zeyde
0f20bfa239 README: add a note about udev 2016-12-06 18:36:14 +02:00
Roman Zeyde
798597c436 bump version 2016-12-06 10:19:23 +02:00
Roman Zeyde
a13b1103f7 setup: temporary pin dependency for keepkey compatibilty 2016-12-06 10:07:18 +02:00
Roman Zeyde
9fe1a235c1 gpg: check that the configuration is in place 2016-12-02 13:10:33 +02:00
Roman Zeyde
f86aae9a40 gpg: check that GnuPG v2 is installed 2016-12-02 12:48:56 +02:00
Roman Zeyde
fc070e3ca0 GPG: add OS X installation link to README 2016-11-30 23:08:56 +02:00
Roman Zeyde
05fac995eb README: add 'setup.py' installation 2016-11-26 14:01:06 +02:00
Roman Zeyde
188b74b327 gpg: use explicit '--subkey' flag for adding a subkey to an existing GPG key 2016-11-25 19:35:40 +02:00
Roman Zeyde
fc31847f8e decode: add test for custom markers 2016-11-19 20:06:29 +02:00
Roman Zeyde
0faf21a102 README: easier tag signature verification 2016-11-19 13:37:33 +02:00
Roman Zeyde
6b82f8b9b7 keyring: add test for get_agent_sock_path() 2016-11-12 20:51:35 +02:00
Roman Zeyde
fabfcaaae2 keyring: fix test case for iterlines() 2016-11-12 20:51:14 +02:00
Roman Zeyde
f0f89310ac main: add '--mosh' for better SSH client 2016-11-11 22:26:22 +02:00
Roman Zeyde
47ff7c5cb3 gpg: add usage example for GPA 2016-11-11 20:05:47 +02:00
Roman Zeyde
0440025083 gpg: use explicit function to check for custom subpacket marker 2016-11-11 13:02:02 +02:00
Roman Zeyde
c49fe97f63 gpg: remove unused parser for literal packets 2016-11-11 13:01:54 +02:00
Roman Zeyde
7f8abcb5c5 client: remove unused code 2016-11-11 13:01:47 +02:00
Roman Zeyde
e13039e52d gpg: remove property method and unused member variable from PublicKey 2016-11-11 13:01:33 +02:00
Roman Zeyde
c420571eb8 gpg: import test coverage for protocol 2016-11-11 09:14:33 +02:00
Roman Zeyde
827119a18d gpg: handle KILLAGENT command
so `gpg-connect-agent KILLAGENT` should stop the running agent
2016-11-10 23:29:47 +02:00
Roman Zeyde
9be6504658 util: import test coverage 2016-11-10 14:33:41 +02:00
Roman Zeyde
07cbe65875 formats: improve test coverage 2016-11-10 14:33:27 +02:00
Roman Zeyde
180120e787 README: link to trezor-ssh-agent for Windows support 2016-11-10 13:09:53 +02:00
Roman Zeyde
f4ce81fa94 README: add 1.4.0-related post 2016-11-10 13:03:12 +02:00
Roman Zeyde
176bf4ef7c README: add Hg demo 2016-11-08 21:45:42 +02:00
Roman Zeyde
d22cd7512d gpg: launch agent before starting the shell 2016-11-06 22:39:45 +02:00
Roman Zeyde
83f17704cb server: remove 'SSH_AUTH_SOCK=' from logging 2016-11-06 22:02:35 +02:00
Roman Zeyde
92f6751ccb README: multiple SSH identities from config file 2016-11-06 20:46:42 +02:00
Roman Zeyde
abe80533eb README: trim whitespace 2016-11-06 20:43:45 +02:00
Roman Zeyde
de51665c71 bump version 2016-11-04 21:16:15 +02:00
Roman Zeyde
c30e5f5a67 gpg: add optional arguments to gpg-shell 2016-11-04 19:33:05 +02:00
Roman Zeyde
2eab2a152c device: verify keepkey constraints 2016-11-04 19:31:54 +02:00
Roman Zeyde
5e93d97be3 Merge branch 'ssh-ids' 2016-11-04 16:07:29 +02:00
Roman Zeyde
4c8fcd6714 ssh: use special UNIX socket name 2016-11-04 10:36:53 +02:00
Roman Zeyde
ee593bc66e gpg: show user ID on a single line 2016-11-03 23:36:11 +02:00
Roman Zeyde
dbed773e54 fix pylint and tests 2016-11-03 23:29:45 +02:00
Roman Zeyde
ac4a86d312 ssh: remove git utility 2016-11-03 23:12:59 +02:00
Roman Zeyde
021831073e ssh: simple support for multiple public keys loading 2016-11-03 23:05:27 +02:00
Roman Zeyde
6a5acba0b0 gpg: decouple identity from device 2016-11-03 22:00:56 +02:00
Roman Zeyde
9123cef810 ssh: decouple identity from device 2016-11-03 22:00:43 +02:00
Roman Zeyde
6f6e7c0bcc device: allow loading identities from a file (instead of argument) 2016-11-03 22:00:22 +02:00
Roman Zeyde
47ff081525 bump version 2016-11-03 16:46:14 +02:00
Roman Zeyde
6d53baafe2 setup: add 'device' package 2016-11-03 16:45:39 +02:00
Roman Zeyde
317b672add setup: support Python 3.5 2016-10-31 23:00:19 +02:00
Roman Zeyde
9964c200ff travis: install keepkey from GitHub 2016-10-31 22:40:23 +02:00
Roman Zeyde
75405b4944 gpg: allow PIN entry before starting GPG shell 2016-10-30 22:03:39 +02:00
Roman Zeyde
e74b9c77af gpg: rename gpg.device into gpg.client 2016-10-30 22:03:12 +02:00
Roman Zeyde
c2158947c8 Merge branch 'refactor-device' 2016-10-30 20:29:44 +02:00
Roman Zeyde
e39d5025d5 travis: use pip to install trezor-agent 2016-10-30 20:23:20 +02:00
Roman Zeyde
efdb9fcfb5 gpg: fix bytes/str issue with GPG user ID 2016-10-30 20:23:20 +02:00
Roman Zeyde
a20b1ed2a8 factory: remove obsolete code 2016-10-30 20:23:20 +02:00
Roman Zeyde
ca507126d6 gpg: use new device package (instead of factory) 2016-10-30 20:23:20 +02:00
Roman Zeyde
0f79b5ff2e ssh: use new device package (instead of factory) 2016-10-30 20:23:19 +02:00
Roman Zeyde
946ab633d4 device: move device-related code into a separate package 2016-10-30 20:23:19 +02:00
Roman Zeyde
4108c9287f bump version 2016-10-29 22:50:43 +03:00
Roman Zeyde
d9cb75e95d gpg: use which gpg2 for 'gpg.program' git configuration
This enables git-cola support TREZOR-based commit signatues,
together with "cola.signcommits=true" setting.
2016-10-29 22:28:57 +03:00
Roman Zeyde
2cecd2ed08 pylint: enable "fixme" 2016-10-29 19:44:54 +03:00
Roman Zeyde
05f40085b2 gpg: remove spurious space 2016-10-29 17:25:22 +03:00
Roman Zeyde
c7346d621d gpg: use policy URI subpacket for marking our public keys
keybase.io does not support experimental/private subpacket IDs
2016-10-29 17:16:36 +03:00
Roman Zeyde
0342b39465 gpg: allow setting key creation timestamp 2016-10-29 17:16:20 +03:00
Roman Zeyde
fa6d8564b9 gpg: use SHA-512 as default hash for signatures 2016-10-29 11:25:53 +03:00
Roman Zeyde
e09712c793 gpg: add OS X support for gpg-init 2016-10-25 19:55:48 +03:00
Roman Zeyde
0cbb3bb9fa Merge pull request #67 from romanz/concurrent-handler
Concurrent SSH handler
2016-10-24 21:51:19 +03:00
Roman Zeyde
d7a6641ffa gpg: update screencasts 2016-10-24 20:13:54 +03:00
Roman Zeyde
6fe89241c4 Merge branch 'auto-spawn-agent' 2016-10-24 19:29:45 +03:00
Roman Zeyde
c5262d075b gpg: use 'gpg-agent.conf' to configure trezor-gpg-agent
currently support logfile and logging verbosity
2016-10-24 17:55:35 +03:00
Roman Zeyde
683d24f4eb gpg: use gpg.conf to automatically spawn trezor-gpg-agent 2016-10-24 17:54:39 +03:00
Roman Zeyde
921e2954c1 gpg: support more digests (with larger output than 256 bits)
NIST256 signs the prefix of a longer digest.
Ed25519 signs the whole one.
2016-10-24 16:41:12 +03:00
Roman Zeyde
3f784289d8 gpg: allow setting CURVE from environment 2016-10-24 11:01:37 +03:00
Roman Zeyde
04d790767d gpg: don't fail on non-zero shell exit code 2016-10-23 21:37:09 +03:00
Roman Zeyde
97efdf4a45 ssh: handle connections concurrently 2016-10-23 17:35:12 +03:00
Roman Zeyde
ee2f6b75dc server: log SSH version for debugging 2016-10-23 17:05:20 +03:00
Roman Zeyde
a26f0ea034 README: make tag example clearer 2016-10-23 13:56:22 +03:00
Roman Zeyde
a68f1e5c26 gpg: update README for easier usage 2016-10-22 22:46:18 +03:00
Roman Zeyde
93e3c66a15 gpg: notify the user for confirmation 2016-10-22 22:35:49 +03:00
Roman Zeyde
44eaaa6b9c gpg: don't spawn gpg-shell automatically 2016-10-22 22:35:08 +03:00
Roman Zeyde
b83d4960e7 gpg: run gpg-shell in verbose mode 2016-10-22 22:34:55 +03:00
Roman Zeyde
75fe7b4e05 gpg: improve shell helper scripts
- explicit trust configuration
- less debug prints
2016-10-22 21:41:23 +03:00
Roman Zeyde
742136b22d gpg: add helper scripts 2016-10-21 23:19:32 +03:00
Roman Zeyde
513e99dd57 server: refactor server_thread() to decouple it from handle_connection() 2016-10-21 22:09:47 +03:00
Roman Zeyde
1bd6775c35 gpg: replace -s flag by implicit adding to existing GPG key 2016-10-21 21:25:22 +03:00
Roman Zeyde
aaade1737f gpg: comment about digest size 2016-10-21 19:02:47 +03:00
Roman Zeyde
fe185c190e ledger: move factory-related code to a separate file 2016-10-21 18:34:36 +03:00
Roman Zeyde
1bc0165368 setup: update trezorlib dependency 2016-10-21 18:07:04 +03:00
Roman Zeyde
0f841ffbc4 factory: add Python 3 support for Ledger 2016-10-21 17:33:26 +03:00
Roman Zeyde
b2942035a3 gpg: skip "progress" status messages 2016-10-20 22:46:39 +03:00
Roman Zeyde
215b64f253 gpg: fix comment 2016-10-18 22:23:40 +03:00
Roman Zeyde
79e68b29c2 bump version 2016-10-18 22:16:54 +03:00
Roman Zeyde
8265515641 gpg: fix small Python2/3 issue 2016-10-18 22:15:58 +03:00
Roman Zeyde
749799845d bump version 2016-10-18 21:37:01 +03:00
Roman Zeyde
eaea35003e gpg: remove unused function (_time_format) 2016-10-18 21:25:31 +03:00
Roman Zeyde
eefb38ce83 gpg: remove unused function (_verify_keygrip) 2016-10-18 21:19:09 +03:00
Roman Zeyde
0730eb7223 gpg: use same logging configuration as in SSH 2016-10-18 21:02:49 +03:00
Roman Zeyde
5b61702205 gpg: don't crash gpg-agent on error 2016-10-18 20:56:17 +03:00
Roman Zeyde
0ad0ca3b9a README: add a note about SSH incompatible options 2016-10-18 19:46:43 +03:00
Roman Zeyde
2843cdcf41 ssh: pretty-print user name 2016-10-18 18:28:21 +03:00
Roman Zeyde
c7bc78ebe7 Merge pull request #58 from romanz/keygrip-agent
gpg: replace TREZOR_GPG_USER_ID usage in gpg-agent mode
2016-10-18 18:15:00 +03:00
Roman Zeyde
a6d9edcb0b README: update for new user ID specification for GPG 2016-10-18 18:12:42 +03:00
Roman Zeyde
bc64205a85 gpg: replace TREZOR_GPG_USER_ID usage in gpg-agent mode
Use the keygrip to find the correct public key instead.
2016-10-18 18:05:51 +03:00
Roman Zeyde
34dc803856 README: add "user@" for SSH example usage
This should help when local username and remote username are different.
2016-10-18 15:07:40 +03:00
Roman Zeyde
f7ebb02799 isort: fix imports 2016-10-18 12:10:28 +03:00
Roman Zeyde
0ba33a5bc4 gpg: document agent responses 2016-10-18 12:08:28 +03:00
Roman Zeyde
13752ddcd5 gpg: require latest GPG version 2016-10-18 12:05:44 +03:00
Roman Zeyde
487a8e56c4 gpg: add keygrip logic into decoding 2016-10-17 23:30:50 +03:00
Roman Zeyde
ef56ee4602 gpg: remove verifying logic from decoding 2016-10-17 23:08:16 +03:00
Roman Zeyde
ae381a38e5 gpg: export keygrips from protocol 2016-10-17 22:57:40 +03:00
Roman Zeyde
446ec99bf4 gpg: remove complex pubkey parsing code 2016-10-17 22:51:11 +03:00
Roman Zeyde
80c6f10533 README: correct pip commands order 2016-10-17 18:01:38 +03:00
Roman Zeyde
ff984c60e4 README: link to PIN entering instructions 2016-10-17 17:54:59 +03:00
Roman Zeyde
c9bc079dc9 gpg: add file:line to logging format 2016-10-17 11:58:03 +03:00
Roman Zeyde
65d2c04478 gpg: fix agent module to work with Python 3 2016-10-17 11:47:22 +03:00
Roman Zeyde
2d57bf4453 gpg: beter logging while search for GPG key 2016-10-17 11:46:58 +03:00
Roman Zeyde
79b6d31dfe gpg: raise proper exception when keygrip mismatch is detected 2016-10-17 11:08:06 +03:00
Roman Zeyde
7de88a3980 gpg: add comment for stopping current gpg-agent 2016-10-16 22:40:16 +03:00
Roman Zeyde
6f8d0df116 bump version 2016-10-16 22:17:53 +03:00
Roman Zeyde
b4a382d22e Merge pull request #51 from romanz/curve25519
Curve25519
2016-10-15 16:19:50 +03:00
Roman Zeyde
d236f4667e gpg: allow Curve25519 for ECDH 2016-10-15 16:10:16 +03:00
Roman Zeyde
42813ddbb4 gpg: parse curve OID from public key to select curve name 2016-10-15 16:10:16 +03:00
Roman Zeyde
8f19690943 gpg: support Curve25519 for creating encryption subkeys 2016-10-15 16:10:16 +03:00
Roman Zeyde
5047805385 gpg: move HardwareSigner to device module 2016-10-15 16:10:16 +03:00
Roman Zeyde
915b326da7 gpg: simplify AgentSigner and move to keyring module 2016-10-15 15:57:45 +03:00
Roman Zeyde
e7b8379a97 factory: explicitly only the first interface 2016-10-14 20:58:42 +03:00
Roman Zeyde
26435130d7 factory: emit warning (instead of exception) when an import fails 2016-10-12 21:15:21 +03:00
Roman Zeyde
dfde6dbee4 bump version 2016-10-12 19:12:04 +03:00
Cédric Félizard
085a3e81c7 Add explanation about entering PIN (#47)
[ci skip]
2016-10-11 21:31:25 +03:00
Cédric Félizard
3082d61deb Fix typo (#48) 2016-10-11 21:29:54 +03:00
Roman Zeyde
e3286a4510 gpg: don't clear the session after PIN is entered
This would allow single PIN entry when running multiple GPG commands.
2016-10-11 08:43:39 +03:00
Roman Zeyde
fcd5671626 Handle keyinfo request (#44)
gpg: handle KEYINFO request

See https://git.gnupg.org/cgi-bin/gitweb.cgi?p=gnupg.git;a=blob;f=agent/command.c;h=9522f898997e95207d59122d056f0f0be03ccecb;hb=6bee88dd067e03e7767ceacf6a849d9ba38cc11d#l1027 for more details.
2016-10-04 23:11:12 +03:00
Roman Zeyde
1454d2f4d7 Merge pull request #43 from romanz/fix-gpg-manual
GPG: fix installation instructions
2016-10-04 20:01:44 +03:00
Roman Zeyde
9b395363a3 GPG: fix installation instructions 2016-10-04 19:59:08 +03:00
Roman Zeyde
5bb9dd7770 Merge pull request #42 from romanz/ssh-git
README: add an example for remote git repository
2016-10-04 11:43:55 +03:00
Roman Zeyde
51df023a23 README: add an example for remote git repository 2016-10-04 11:40:14 +03:00
Roman Zeyde
d74f375637 Merge pull request #41 from romanz/protobuf-note
README: add note about protobuf permissions issue
2016-10-04 11:27:14 +03:00
Roman Zeyde
1fd0659051 README: add note about protobuf permissions issue 2016-10-04 11:24:39 +03:00
Roman Zeyde
18be290bd6 Merge branch 'fix_agent' of https://github.com/Solution4Future/trezor-agent into Solution4Future-fix_agent 2016-10-04 11:16:56 +03:00
Roman Zeyde
a1ab496bf4 Merge branch 'ledger' 2016-10-04 10:39:08 +03:00
Roman Zeyde
784e14647a Merge branch 'master' into HEAD
Conflicts:
	trezor_agent/factory.py
2016-10-04 10:37:52 +03:00
Dominik Kozaczko
7d2c649e83 don't stop polling for more devices as having more than one inserted raises more problems and we need to keep the check 2016-10-01 12:38:16 +02:00
Dominik Kozaczko
cf27b345f6 better handling of keepkey dependency; fixes #36 2016-10-01 12:30:00 +02:00
Dominik Kozaczko
386ed5a81f Merge pull request #1 from romanz/master
pull changes from upstream
2016-10-01 12:10:18 +02:00
Roman Zeyde
5a64954324 Merge pull request #37 from Solution4Future/fix_agent
removed .decode('ascii') and added missing bytestrings and also some missing deps
2016-10-01 11:42:47 +03:00
Dominik Kozaczko
3aebd137b0 removed .decode('ascii') and added missing bytestrings 2016-10-01 10:02:46 +02:00
Nicolas Pouillard
1fa35e7f1a Fix the URL for the TREZOR firmware 2016-09-30 21:07:35 +03:00
Roman Zeyde
aeda85275d bump version 2016-09-28 18:32:19 +03:00
Roman Zeyde
e41206b350 setup: update trezorlib version 2016-09-28 18:31:13 +03:00
Roman Zeyde
03650550dd Use latest protobuf library (for native Python 3 support) 2016-09-28 18:18:09 +03:00
Roman Zeyde
f7b07070da README: update setuptools to the latest version 2016-09-28 18:08:09 +03:00
Roman Zeyde
96eede9c83 Merge branch 'np-encode-subpackets' 2016-09-28 17:27:48 +03:00
Roman Zeyde
91146303a3 Follow GPG implementation for subpacket prefix encoding.
Conflicts:
	trezor_agent/gpg/protocol.py
2016-09-28 17:26:50 +03:00
Roman Zeyde
bf598435fb client: keep the session open (doesn't forget PIN) 2016-09-26 22:27:47 +03:00
Roman Zeyde
459b882b89 ledger: don't use debug=True 2016-09-14 23:07:27 +03:00
Roman Zeyde
998c9ee958 README: update usage section 2016-09-11 23:38:21 +03:00
Roman Zeyde
d408a592aa README: get only the first lines of 'trezorctl get_features' 2016-09-11 23:35:10 +03:00
Roman Zeyde
282e91ace3 update README about protobuf issueOF 2016-09-11 23:22:10 +03:00
Roman Zeyde
23c37cf1e3 README: update TREZOR's required version 2016-09-11 23:17:47 +03:00
Nicolas Pouillard
016e864503 Attempt at fixing issue #32 2016-09-06 00:45:51 +02:00
Roman Zeyde
57e09248db Merge pull request #31 from romanz/master
Update ledger branch with the latest changes from master branch
2016-09-05 22:28:07 +03:00
Roman Zeyde
030ae4c3f6 gpg: include unsupport hash algorithm ID in exception message 2016-08-13 10:06:52 +03:00
Roman Zeyde
4897b70888 factory: fix pylint import-error warnings 2016-08-11 22:38:12 +03:00
Roman Zeyde
f4ecd47ed6 factory: fix pep8 and pylint warnings 2016-08-11 22:31:24 +03:00
Roman Zeyde
c4bbac0e77 util: move BIP32 address related functions 2016-08-11 22:30:59 +03:00
BTChip
5d0b0f65d3 Merge branch 'ledger' of https://github.com/btchip/trezor-agent into ledger 2016-08-09 13:02:47 +02:00
BTChip
33747592ca Fix eddsa, SSH optimization with signature + key, cleanup 2016-08-09 13:01:57 +02:00
BTChip
adb09cd8ca Ledger integration 2016-08-09 13:01:57 +02:00
BTChip
bc1d7a5448 Fix eddsa, SSH optimization with signature + key, cleanup 2016-08-03 15:17:04 +02:00
BTChip
8fe16d24c2 Ledger integration 2016-07-31 10:00:46 +02:00
39 changed files with 1245 additions and 1067 deletions

View File

@@ -1,2 +1,2 @@
[MESSAGES CONTROL]
disable=fixme, invalid-name, missing-docstring, locally-disabled, unbalanced-tuple-unpacking
disable=invalid-name, missing-docstring, locally-disabled, unbalanced-tuple-unpacking

View File

@@ -5,9 +5,22 @@ python:
- "3.4"
- "3.5"
cache:
directories:
- $HOME/.cache/pip
addons:
apt:
packages:
- libudev-dev
- libusb-1.0-0-dev
before_install:
- pip install -U setuptools pylint coverage pep8 pydocstyle "pip>=7.0" wheel
- pip install -e git+https://github.com/keepkey/python-keepkey@6e8baa8b935e830d05f87b6dfd9bc7c927a96dc3#egg=keepkey
install:
- pip install ecdsa ed25519 semver # test without trezorlib for now
- pip install -U pylint coverage pep8 pydocstyle # use latest tools
- pip install -e .
script:
- pep8 trezor_agent

View File

@@ -4,101 +4,48 @@ Thanks!
# Installation
First, verify that you have GPG 2.1+ [installed](https://gist.github.com/vt0r/a2f8c0bcb1400131ff51):
First, verify that you have GPG 2.1.11+ installed
([Debian](https://gist.github.com/vt0r/a2f8c0bcb1400131ff51),
[macOS](https://sourceforge.net/p/gpgosx/docu/Download/)):
```
$ gpg2 --version | head -n1
gpg (GnuPG) 2.1.11
gpg (GnuPG) 2.1.15
```
Update you TREZOR firmware to the latest version (at least [c720614](https://github.com/trezor/trezor-mcu/commit/c720614f6e9b9c07f446c95bda0257980d942871)).
This GPG version is included in [Ubuntu 16.04](https://launchpad.net/ubuntu/+source/gnupg2)
and [Linux Mint 18](https://community.linuxmint.com/software/view/gnupg2).
Install latest `trezor-agent` package from [gpg-agent](https://github.com/romanz/trezor-agent/commits/gpg-agent) branch:
Update you TREZOR firmware to the latest version (at least v1.4.0).
Install latest `trezor-agent` package from GitHub:
```
$ pip install --user git+https://github.com/romanz/trezor-agent.git
```
Define your GPG user ID as an environment variable:
```
$ export TREZOR_GPG_USER_ID="John Doe <john@doe.bit>"
```
# Quickstart
There are two ways to generate TREZOR-based GPG public keys, as described below.
## Identity creation
[![asciicast](https://asciinema.org/a/c2yodst21h9obttkn9wgf3783.png)](https://asciinema.org/a/c2yodst21h9obttkn9wgf3783)
## 1. generate a new GPG identity:
## Sample usage (signature and decryption)
[![asciicast](https://asciinema.org/a/7x0h9tyoyu5ar6jc8y9oih0ba.png)](https://asciinema.org/a/7x0h9tyoyu5ar6jc8y9oih0ba)
You can use GNU Privacy Assistant (GPA) in order to inspect the created keys
and perform signature and decryption operations using:
```
$ trezor-gpg create | gpg2 --import # use the TREZOR to confirm signing the primary key
gpg: key 5E4D684D: public key "John Doe <john@doe.bit>" imported
gpg: Total number processed: 1
gpg: imported: 1
$ gpg2 --edit "${TREZOR_GPG_USER_ID}" trust # set this key to ultimate trust (option #5)
$ gpg2 -k
/home/roman/.gnupg/pubring.kbx
------------------------------
pub nistp256/5E4D684D 2016-06-17 [SC]
uid [ultimate] John Doe <john@doe.bit>
sub nistp256/A31D9E25 2016-06-17 [E]
```
## 2. generate a new subkey for an existing GPG identity:
```
$ gpg2 -k # suppose there is already a GPG primary key
/home/roman/.gnupg/pubring.kbx
------------------------------
pub rsa2048/87BB07B4 2016-06-17 [SC]
uid [ultimate] John Doe <john@doe.bit>
sub rsa2048/7176D31F 2016-06-17 [E]
$ trezor-gpg create --subkey | gpg2 --import # use the TREZOR to confirm signing the subkey
gpg: key 87BB07B4: "John Doe <john@doe.bit>" 2 new signatures
gpg: key 87BB07B4: "John Doe <john@doe.bit>" 2 new subkeys
gpg: Total number processed: 1
gpg: new subkeys: 2
gpg: new signatures: 2
$ gpg2 -k
/home/roman/.gnupg/pubring.kbx
------------------------------
pub rsa2048/87BB07B4 2016-06-17 [SC]
uid [ultimate] John Doe <john@doe.bit>
sub rsa2048/7176D31F 2016-06-17 [E]
sub nistp256/DDE80B36 2016-06-17 [S]
sub nistp256/E3D0BA19 2016-06-17 [E]
```
# Usage examples:
## Start the TREZOR-based gpg-agent:
```
$ trezor-gpg agent &
```
Note: this agent intercepts all GPG requests, so make sure to close it (e.g. by using `killall trezor-gpg`),
when you are done with the TREZOR-based GPG operations.
## Sign and verify GPG messages:
```
$ echo "Hello World!" | gpg2 --sign | gpg2 --verify
gpg: Signature made Fri 17 Jun 2016 08:55:13 PM IDT using ECDSA key ID 5E4D684D
gpg: Good signature from "Roman Zeyde <roman.zeyde@gmail.com>" [ultimate]
```
## Encrypt and decrypt GPG messages:
```
$ date | gpg2 --encrypt -r "${TREZOR_GPG_USER_ID}" | gpg2 --decrypt
gpg: encrypted with 256-bit ECDH key, ID A31D9E25, created 2016-06-17
"Roman Zeyde <roman.zeyde@gmail.com>"
Fri Jun 17 20:55:31 IDT 2016
$ sudo apt install gpa
$ ./scripts/gpg-shell gpa
```
[![GPA](https://cloud.githubusercontent.com/assets/9900/20224804/053d7474-a849-11e6-87f3-ab07dc536158.png)](https://www.gnupg.org/related_software/swlist.html#gpa)
## Git commit & tag signatures:
Git can use GPG to sign and verify commits and tags (see [here](https://git-scm.com/book/en/v2/Git-Tools-Signing-Your-Work)):
```
$ git config --local gpg.program gpg2
$ git config --local gpg.program $(which gpg2)
$ git commit --gpg-sign # create GPG-signed commit
$ git log --show-signature -1 # verify commit signature
$ git tag --sign "TAG" # create GPG-signed tag
$ git verify-tag "TAG" # verify tag signature
```
$ git tag v1.2.3 --sign # create GPG-signed tag
$ git tag v1.2.3 --verify # verify tag signature
```

View File

@@ -9,38 +9,20 @@
## Using for GitHub SSH authentication (via `trezor-git` utility)
[![GitHub](https://asciinema.org/a/38337.png)](https://asciinema.org/a/38337)
# Installation
First, make sure that the latest [trezorlib](https://pypi.python.org/pypi/trezor) Python package
is installed correctly (at least v0.6.6):
$ apt-get install python-dev libusb-1.0-0-dev libudev-dev
$ pip install Cython trezor
Then, install the latest [trezor_agent](https://pypi.python.org/pypi/trezor_agent) package:
$ pip install trezor_agent
Finally, verify that you are running the latest [TREZOR firmware](https://mytrezor.com/data/firmware/releases.json) version (at least v1.3.4):
$ trezorctl get_features
vendor: "bitcointrezor.com"
major_version: 1
minor_version: 3
patch_version: 4
...
## Loading multiple SSH identities from configuration file
[![Config](https://asciinema.org/a/bdxxtgctk5syu56yfz8lcp7ny.png)](https://asciinema.org/a/bdxxtgctk5syu56yfz8lcp7ny)
# Public key generation
Run:
/tmp $ trezor-agent ssh.hostname.com -v > hostname.pub
2015-09-02 15:03:18,929 INFO getting "ssh://ssh.hostname.com" public key from Trezor...
/tmp $ trezor-agent user@ssh.hostname.com -v > hostname.pub
2015-09-02 15:03:18,929 INFO getting "ssh://user@ssh.hostname.com" public key from Trezor...
2015-09-02 15:03:23,342 INFO disconnected from Trezor
/tmp $ cat hostname.pub
ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBGSevcDwmT+QaZPUEWUUjTeZRBICChxMKuJ7dRpBSF8+qt+8S1GBK5Zj8Xicc8SHG/SE/EXKUL2UU3kcUzE7ADQ= ssh://ssh.hostname.com
ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBGSevcDwmT+QaZPUEWUUjTeZRBICChxMKuJ7dRpBSF8+qt+8S1GBK5Zj8Xicc8SHG/SE/EXKUL2UU3kcUzE7ADQ= ssh://user@ssh.hostname.com
Append `hostname.pub` contents to `~/.ssh/authorized_keys`
Append `hostname.pub` contents to `/home/user/.ssh/authorized_keys`
configuration file at `ssh.hostname.com`, so the remote server
would allow you to login using the corresponding private key signature.
@@ -48,9 +30,9 @@ would allow you to login using the corresponding private key signature.
Run:
/tmp $ trezor-agent ssh.hostname.com -v -c
2015-09-02 15:09:39,782 INFO getting "ssh://ssh.hostname.com" public key from Trezor...
2015-09-02 15:09:44,430 INFO please confirm user "roman" login to "ssh://ssh.hostname.com" using Trezor...
/tmp $ trezor-agent user@ssh.hostname.com -v -c
2015-09-02 15:09:39,782 INFO getting "ssh://user@ssh.hostname.com" public key from Trezor...
2015-09-02 15:09:44,430 INFO please confirm user "roman" login to "ssh://user@ssh.hostname.com" using Trezor...
2015-09-02 15:09:46,152 INFO signature status: OK
Linux lmde 3.16.0-4-amd64 #1 SMP Debian 3.16.7-ckt11-1+deb8u3 (2015-08-04) x86_64
@@ -64,3 +46,42 @@ Run:
~ $
Make sure to confirm SSH signature on the Trezor device when requested.
## Accessing remote Git/Mercurial repositories
Use your SSH public key to access your remote repository (e.g. [GitHub](https://help.github.com/articles/adding-a-new-ssh-key-to-your-github-account/)):
$ trezor-agent -v -e ed25519 git@github.com | xclip
Use the following Bash alias for convinient Git operations:
$ alias git_hub='trezor-agent -v -e ed25519 git@github.com -- git'
Replace `git` with `git_hub` for remote operations:
$ git_hub push origin master
The same works for Mercurial (e.g. on [BitBucket](https://confluence.atlassian.com/bitbucket/set-up-ssh-for-mercurial-728138122.html)):
$ trezor-agent -v -e ed25519 git@bitbucket.org -- hg push
# Troubleshooting
If SSH connection fails to work, please open an [issue](https://github.com/romanz/trezor-agent/issues)
with a verbose log attached (by running `trezor-agent -vv`) .
## Incompatible SSH options
Note that your local SSH configuration may ignore `trezor-agent`, if it has `IdentitiesOnly` option set to `yes`.
IdentitiesOnly
Specifies that ssh(1) should only use the authentication identity files configured in
the ssh_config files, even if ssh-agent(1) or a PKCS11Provider offers more identities.
The argument to this keyword must be “yes” or “no”.
This option is intended for situations where ssh-agent offers many different identities.
The default is “no”.
If you are failing to connect, try running:
$ trezor-agent -vv user@host -- ssh -vv -oIdentitiesOnly=no user@host

View File

@@ -10,9 +10,52 @@ See SatoshiLabs' blog posts about this feature:
- [TREZOR Firmware 1.3.4 enables SSH login](https://medium.com/@satoshilabs/trezor-firmware-1-3-4-enables-ssh-login-86a622d7e609)
- [TREZOR Firmware 1.3.6GPG Signing, SSH Login Updates and Advanced Transaction Features for Segwit](https://medium.com/@satoshilabs/trezor-firmware-1-3-6-20a7df6e692)
- [TREZOR Firmware 1.4.0GPG decryption support](https://www.reddit.com/r/TREZOR/comments/50h8r9/new_trezor_firmware_fidou2f_and_initial_ethereum/d7420q7/)
For usage with SSH, see the [following instructions](README-SSH.md).
## Installation
For usage with GPG, see the [following instructions](README-GPG.md).
First, make sure that the latest [trezorlib](https://pypi.python.org/pypi/trezor) Python package
is installed correctly (at least v0.6.6):
$ apt-get install python-dev libusb-1.0-0-dev libudev-dev
$ pip install -U setuptools pip
$ pip install Cython trezor
Make sure that your `udev` rules are configured [correctly](https://doc.satoshilabs.com/trezor-user/settingupchromeonlinux.html#manual-configuration-of-udev-rules).
Then, install the latest [trezor_agent](https://pypi.python.org/pypi/trezor_agent) package:
$ pip install trezor_agent
Or, directly from the latest source code (if `pip` doesn't work for you):
$ git clone https://github.com/romanz/trezor-agent && cd trezor-agent
$ python setup.py build && python setup.py install
Finally, verify that you are running the latest [TREZOR firmware](https://wallet.mytrezor.com/data/firmware/releases.json) version (at least v1.4.0):
$ trezorctl get_features | head
vendor: "bitcointrezor.com"
major_version: 1
minor_version: 4
patch_version: 0
...
If you have an error regarding `protobuf` imports (after installing it), please see [this issue](https://github.com/romanz/trezor-agent/issues/28).
## Usage
For SSH, see the [following instructions](README-SSH.md) (for Windows support,
see [trezor-ssh-agent](https://github.com/martin-lizner/trezor-ssh-agent) project (by Martin Lízner)).
For GPG, see the [following instructions](README-GPG.md).
See [here](https://github.com/romanz/python-trezor#pin-entering) for PIN entering instructions.
## Troubleshooting
If there is an import problem with the installed `protobuf` package,
see [this issue](https://github.com/romanz/trezor-agent/issues/28) for fixing it.
### Gitter
Questions, suggestions and discussions are welcome: [![Chat](https://badges.gitter.im/romanz/trezor-agent.svg)](https://gitter.im/romanz/trezor-agent)

34
scripts/gpg-init Executable file
View File

@@ -0,0 +1,34 @@
#!/bin/bash
set -eu
gpg2 --version >/dev/null # verify that GnuPG 2 is installed
USER_ID="${1}"
HOMEDIR=~/.gnupg/trezor
CURVE=${CURVE:="nist256p1"} # or "ed25519"
TIMESTAMP=${TIMESTAMP:=`date +%s`} # key creation timestamp
# Prepare new GPG home directory for TREZOR-based identity
rm -rf "${HOMEDIR}"
mkdir -p "${HOMEDIR}"
chmod 700 "${HOMEDIR}"
# Generate new GPG identity and import into GPG keyring
trezor-gpg-create -v "${USER_ID}" -t "${TIMESTAMP}" -e "${CURVE}" > "${HOMEDIR}/pubkey.asc"
gpg2 --homedir "${HOMEDIR}" --import < "${HOMEDIR}/pubkey.asc"
rm -f "${HOMEDIR}/S.gpg-agent" # (otherwise, our agent won't be started automatically)
# Make new GPG identity with "ultimate" trust (via its fingerprint)
FINGERPRINT=$(gpg2 --homedir "${HOMEDIR}" --list-public-keys --with-fingerprint --with-colons | sed -n -E 's/^fpr:::::::::([0-9A-F]+):$/\1/p' | head -n1)
echo "${FINGERPRINT}:6" | gpg2 --homedir "${HOMEDIR}" --import-ownertrust
# Prepare GPG configuration file
echo "# TREZOR-based GPG configuration
agent-program $(which trezor-gpg-agent)
personal-digest-preferences SHA512
" | tee "${HOMEDIR}/gpg.conf"
echo "# TREZOR-based GPG agent emulator
log-file ${HOMEDIR}/gpg-agent.log
verbosity 2
" | tee "${HOMEDIR}/gpg-agent.conf"

28
scripts/gpg-shell Executable file
View File

@@ -0,0 +1,28 @@
#!/bin/bash
set -eu
gpg2 --version >/dev/null # verify that GnuPG 2 is installed
export GNUPGHOME=~/.gnupg/trezor
CONFIG_PATH="${GNUPGHOME}/gpg-agent.conf"
if [ ! -f ${CONFIG_PATH} ]
then
echo "No configuration found: ${CONFIG_PATH}"
exit 1
fi
# Make sure that the device is unlocked before starting the shell
trezor-gpg-unlock
# Make sure TREZOR-based gpg-agent is running
gpg-connect-agent --agent-program "$(which trezor-gpg-agent)" </dev/null
COMMAND=$*
if [ -z "${COMMAND}" ]
then
gpg2 --list-public-keys
${SHELL}
else
${COMMAND}
fi

View File

@@ -3,13 +3,17 @@ from setuptools import setup
setup(
name='trezor_agent',
version='0.6.6',
version='0.8.2',
description='Using Trezor as hardware SSH agent',
author='Roman Zeyde',
author_email='roman.zeyde@gmail.com',
url='http://github.com/romanz/trezor-agent',
packages=['trezor_agent', 'trezor_agent.gpg'],
install_requires=['ecdsa>=0.13', 'ed25519>=1.4', 'Cython>=0.23.4', 'protobuf>=2.6.1', 'trezor>=0.6.12', 'semver>=2.2'],
packages=['trezor_agent', 'trezor_agent.device', 'trezor_agent.gpg'],
install_requires=[
'ecdsa>=0.13', 'ed25519>=1.4', 'Cython>=0.23.4', 'protobuf>=3.0.0', 'semver>=2.2',
'trezor>=0.7.6', 'keepkey>=0.7.3', 'ledgerblue>=0.1.8',
'hidapi==0.7.99.post15' # until https://github.com/keepkey/python-keepkey/pull/8 is merged
],
platforms=['POSIX'],
classifiers=[
'Environment :: Console',
@@ -21,6 +25,7 @@ setup(
'Operating System :: POSIX',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3.4',
'Programming Language :: Python :: 3.5',
'Topic :: Software Development :: Libraries :: Python Modules',
'Topic :: System :: Networking',
'Topic :: Communications',
@@ -29,7 +34,8 @@ setup(
],
entry_points={'console_scripts': [
'trezor-agent = trezor_agent.__main__:run_agent',
'trezor-git = trezor_agent.__main__:run_git',
'trezor-gpg = trezor_agent.gpg.__main__:main',
'trezor-gpg-create = trezor_agent.gpg.__main__:main_create',
'trezor-gpg-agent = trezor_agent.gpg.__main__:main_agent',
'trezor-gpg-unlock = trezor_agent.gpg.__main__:auto_unlock',
]},
)

View File

@@ -1,5 +1,5 @@
[tox]
envlist = py27,py34
envlist = py27,py3
[pep8]
max-line-length = 100
[testenv]

View File

@@ -7,14 +7,14 @@ import re
import subprocess
import sys
from . import client, formats, protocol, server
from . import client, device, formats, protocol, server, util
log = logging.getLogger(__name__)
def ssh_args(label):
"""Create SSH command for connecting specified server."""
identity = client.string_to_identity(label, identity_type=dict)
identity = device.interface.string_to_identity(label)
args = []
if 'port' in identity:
@@ -22,7 +22,22 @@ def ssh_args(label):
if 'user' in identity:
args += ['-l', identity['user']]
return ['ssh'] + args + [identity['host']]
return args + [identity['host']]
def mosh_args(label):
"""Create SSH command for connecting specified server."""
identity = device.interface.string_to_identity(label)
args = []
if 'port' in identity:
args += ['-p', identity['port']]
if 'user' in identity:
args += [identity['user']+'@'+identity['host']]
else:
args += [identity['host']]
return args
def create_parser():
@@ -52,6 +67,8 @@ def create_agent_parser():
help='run ${SHELL} as subprocess under SSH agent')
g.add_argument('-c', '--connect', default=False, action='store_true',
help='connect to specified host via SSH')
g.add_argument('--mosh', default=False, action='store_true',
help='connect to specified host via using Mosh')
p.add_argument('identity', type=str, default=None,
help='proto://[user@]host[:port][/path]')
@@ -73,15 +90,6 @@ def create_git_parser():
return p
def setup_logging(verbosity):
"""Configure logging for this tool."""
fmt = ('%(asctime)s %(levelname)-12s %(message)-100s '
'[%(filename)s:%(lineno)d]')
levels = [logging.WARNING, logging.INFO, logging.DEBUG]
level = levels[min(verbosity, len(levels) - 1)]
logging.basicConfig(format=fmt, level=level)
def git_host(remote_name, attributes):
"""Extract git SSH host for specified remote name."""
try:
@@ -102,13 +110,11 @@ def git_host(remote_name, attributes):
return '{user}@{host}'.format(**match.groupdict())
def run_server(conn, public_key, command, debug, timeout):
def run_server(conn, public_keys, command, debug, timeout):
"""Common code for run_agent and run_git below."""
try:
signer = conn.sign_ssh_challenge
public_key = formats.import_public_key(public_key)
log.info('using SSH public key: %s', public_key['fingerprint'])
handler = protocol.Handler(keys=[public_key], signer=signer,
handler = protocol.Handler(keys=public_keys, signer=signer,
debug=debug)
with server.serve(handler=handler, timeout=timeout) as env:
return server.run_process(command=command, environ=env)
@@ -128,58 +134,50 @@ def handle_connection_error(func):
return wrapper
def parse_config(fname):
"""Parse config file into a list of Identity objects."""
contents = open(fname).read()
for identity_str, curve_name in re.findall(r'\<(.*?)\|(.*?)\>', contents):
yield device.interface.Identity(identity_str=identity_str,
curve_name=curve_name)
@handle_connection_error
def run_agent(client_factory=client.Client):
"""Run ssh-agent using given hardware client factory."""
args = create_agent_parser().parse_args()
setup_logging(verbosity=args.verbose)
util.setup_logging(verbosity=args.verbose)
with client_factory(curve=args.ecdsa_curve_name) as conn:
label = args.identity
conn = client_factory(device=device.detect())
if args.identity.startswith('/'):
identities = list(parse_config(fname=args.identity))
else:
identities = [device.interface.Identity(
identity_str=args.identity, curve_name=args.ecdsa_curve_name)]
for index, identity in enumerate(identities):
identity.identity_dict['proto'] = 'ssh'
log.info('identity #%d: %s', index, identity)
public_keys = [conn.get_public_key(i) for i in identities]
if args.connect:
command = ['ssh'] + ssh_args(args.identity) + args.command
elif args.mosh:
command = ['mosh'] + mosh_args(args.identity) + args.command
else:
command = args.command
public_key = conn.get_public_key(label=label)
use_shell = bool(args.shell)
if use_shell:
command = os.environ['SHELL']
if args.connect:
command = ssh_args(label) + args.command
log.debug('SSH connect: %r', command)
if not command:
for pk in public_keys:
sys.stdout.write(pk)
return
use_shell = bool(args.shell)
if use_shell:
command = os.environ['SHELL']
log.debug('using shell: %r', command)
if not command:
sys.stdout.write(public_key)
return
return run_server(conn=conn, public_key=public_key, command=command,
debug=args.debug, timeout=args.timeout)
@handle_connection_error
def run_git(client_factory=client.Client):
"""Run git under ssh-agent using given hardware client factory."""
args = create_git_parser().parse_args()
setup_logging(verbosity=args.verbose)
with client_factory(curve=args.ecdsa_curve_name) as conn:
label = git_host(args.remote, ['pushurl', 'url'])
if not label:
log.error('Could not find "%s" SSH remote in .git/config',
args.remote)
return
public_key = conn.get_public_key(label=label)
if not args.test:
if args.command:
command = ['git'] + args.command
else:
sys.stdout.write(public_key)
return
else:
command = ['ssh', '-T', label]
return run_server(conn=conn, public_key=public_key, command=command,
debug=args.debug, timeout=args.timeout)
public_keys = [formats.import_public_key(pk) for pk in public_keys]
for pk, identity in zip(public_keys, identities):
pk['identity'] = identity
return run_server(conn=conn, public_keys=public_keys, command=command,
debug=args.debug, timeout=args.timeout)

View File

@@ -3,13 +3,10 @@ Connection to hardware authentication device.
It is used for getting SSH public keys and ECDSA signing of server requests.
"""
import binascii
import io
import logging
import re
import struct
from . import factory, formats, util
from . import formats, util
log = logging.getLogger(__name__)
@@ -17,129 +14,36 @@ log = logging.getLogger(__name__)
class Client(object):
"""Client wrapper for SSH authentication device."""
def __init__(self, loader=factory.load, curve=formats.CURVE_NIST256):
def __init__(self, device):
"""Connect to hardware device."""
client_wrapper = loader()
self.client = client_wrapper.connection
self.identity_type = client_wrapper.identity_type
self.device_name = client_wrapper.device_name
self.call_exception = client_wrapper.call_exception
self.curve = curve
self.device = device
def __enter__(self):
"""Start a session, and test connection."""
msg = 'Hello World!'
assert self.client.ping(msg) == msg
return self
def get_public_key(self, identity):
"""Get SSH public key from the device."""
with self.device:
pubkey = self.device.pubkey(identity)
def __exit__(self, *args):
"""Forget PIN, shutdown screen and disconnect."""
log.info('disconnected from %s', self.device_name)
self.client.clear_session()
self.client.close()
vk = formats.decompress_pubkey(pubkey=pubkey,
curve_name=identity.curve_name)
return formats.export_public_key(vk=vk,
label=str(identity))
def get_identity(self, label, index=0):
"""Parse label string into Identity protobuf."""
identity = string_to_identity(label, self.identity_type)
identity.proto = 'ssh'
identity.index = index
return identity
def get_public_key(self, label):
"""Get SSH public key corresponding to specified by label."""
identity = self.get_identity(label=label)
label = identity_to_string(identity) # canonize key label
log.info('getting "%s" public key (%s) from %s...',
label, self.curve, self.device_name)
addr = get_address(identity)
node = self.client.get_public_node(n=addr,
ecdsa_curve_name=self.curve)
pubkey = node.node.public_key
vk = formats.decompress_pubkey(pubkey=pubkey, curve_name=self.curve)
return formats.export_public_key(vk=vk, label=label)
def sign_ssh_challenge(self, label, blob):
"""Sign given blob using a private key, specified by the label."""
identity = self.get_identity(label=label)
def sign_ssh_challenge(self, blob, identity):
"""Sign given blob using a private key on the device."""
msg = _parse_ssh_blob(blob)
log.debug('%s: user %r via %r (%r)',
msg['conn'], msg['user'], msg['auth'], msg['key_type'])
log.debug('nonce: %s', binascii.hexlify(msg['nonce']))
log.debug('fingerprint: %s', msg['public_key']['fingerprint'])
log.debug('nonce: %r', msg['nonce'])
fp = msg['public_key']['fingerprint']
log.debug('fingerprint: %s', fp)
log.debug('hidden challenge size: %d bytes', len(blob))
log.info('please confirm user "%s" login to "%s" using %s...',
msg['user'], label, self.device_name)
msg['user'].decode('ascii'), identity,
self.device)
try:
result = self.client.sign_identity(identity=identity,
challenge_hidden=blob,
challenge_visual='',
ecdsa_curve_name=self.curve)
except self.call_exception as e:
code, msg = e.args
log.warning('%s error #%s: %s', self.device_name, code, msg)
raise IOError(msg) # close current connection, keep server open
verifying_key = formats.decompress_pubkey(pubkey=result.public_key,
curve_name=self.curve)
key_type, blob = formats.serialize_verifying_key(verifying_key)
assert blob == msg['public_key']['blob']
assert key_type == msg['key_type']
assert len(result.signature) == 65
assert result.signature[:1] == bytearray([0])
return result.signature[1:]
_identity_regexp = re.compile(''.join([
'^'
r'(?:(?P<proto>.*)://)?',
r'(?:(?P<user>.*)@)?',
r'(?P<host>.*?)',
r'(?::(?P<port>\w*))?',
r'(?P<path>/.*)?',
'$'
]))
def string_to_identity(s, identity_type):
"""Parse string into Identity protobuf."""
m = _identity_regexp.match(s)
result = m.groupdict()
log.debug('parsed identity: %s', result)
kwargs = {k: v for k, v in result.items() if v}
return identity_type(**kwargs)
def identity_to_string(identity):
"""Dump Identity protobuf into its string representation."""
result = []
if identity.proto:
result.append(identity.proto + '://')
if identity.user:
result.append(identity.user + '@')
result.append(identity.host)
if identity.port:
result.append(':' + identity.port)
if identity.path:
result.append(identity.path)
return ''.join(result)
def get_address(identity, ecdh=False):
"""Compute BIP32 derivation address according to SLIP-0013/0017."""
index = struct.pack('<L', identity.index)
addr = index + identity_to_string(identity).encode('ascii')
log.debug('address string: %r', addr)
digest = formats.hashfunc(addr).digest()
s = io.BytesIO(bytearray(digest))
hardened = 0x80000000
addr_0 = [13, 17][bool(ecdh)]
address_n = [addr_0] + list(util.recv(s, '<LLLL'))
return [(hardened | value) for value in address_n]
with self.device:
return self.device.sign(blob=blob, identity=identity)
def _parse_ssh_blob(data):

View File

@@ -0,0 +1,27 @@
"""Cryptographic hardware device management."""
import logging
from . import trezor
from . import keepkey
from . import ledger
from . import interface
log = logging.getLogger(__name__)
DEVICE_TYPES = [
trezor.Trezor,
keepkey.KeepKey,
ledger.LedgerNanoS,
]
def detect():
"""Detect the first available device and return it to the user."""
for device_type in DEVICE_TYPES:
try:
with device_type() as d:
return d
except interface.NotFoundError as e:
log.debug('device not found: %s', e)
raise IOError('No device found!')

View File

@@ -0,0 +1,135 @@
"""Device abstraction layer."""
import hashlib
import io
import logging
import re
import struct
from .. import formats, util
log = logging.getLogger(__name__)
_identity_regexp = re.compile(''.join([
'^'
r'(?:(?P<proto>.*)://)?',
r'(?:(?P<user>.*)@)?',
r'(?P<host>.*?)',
r'(?::(?P<port>\w*))?',
r'(?P<path>/.*)?',
'$'
]))
def string_to_identity(identity_str):
"""Parse string into Identity dictionary."""
m = _identity_regexp.match(identity_str)
result = m.groupdict()
log.debug('parsed identity: %s', result)
return {k: v for k, v in result.items() if v}
def identity_to_string(identity_dict):
"""Dump Identity dictionary into its string representation."""
result = []
if identity_dict.get('proto'):
result.append(identity_dict['proto'] + '://')
if identity_dict.get('user'):
result.append(identity_dict['user'] + '@')
result.append(identity_dict['host'])
if identity_dict.get('port'):
result.append(':' + identity_dict['port'])
if identity_dict.get('path'):
result.append(identity_dict['path'])
log.debug('identity parts: %s', result)
return ''.join(result)
class Error(Exception):
"""Device-related error."""
class NotFoundError(Error):
"""Device could not be found."""
class DeviceError(Error):
""""Error during device operation."""
class Identity(object):
"""Represent SLIP-0013 identity, together with a elliptic curve choice."""
def __init__(self, identity_str, curve_name):
"""Configure for specific identity and elliptic curve usage."""
self.identity_dict = string_to_identity(identity_str)
self.curve_name = curve_name
def items(self):
"""Return a copy of identity_dict items."""
return self.identity_dict.items()
def __str__(self):
"""Return identity serialized to string."""
return '<{}|{}>'.format(identity_to_string(self.identity_dict), self.curve_name)
def get_bip32_address(self, ecdh=False):
"""Compute BIP32 derivation address according to SLIP-0013/0017."""
index = struct.pack('<L', self.identity_dict.get('index', 0))
addr = index + identity_to_string(self.identity_dict).encode('ascii')
log.debug('bip32 address string: %r', addr)
digest = hashlib.sha256(addr).digest()
s = io.BytesIO(bytearray(digest))
hardened = 0x80000000
addr_0 = [13, 17][bool(ecdh)]
address_n = [addr_0] + list(util.recv(s, '<LLLL'))
return [(hardened | value) for value in address_n]
def get_curve_name(self, ecdh=False):
"""Return correct curve name for device operations."""
if ecdh:
return formats.get_ecdh_curve_name(self.curve_name)
else:
return self.curve_name
class Device(object):
"""Abstract cryptographic hardware device interface."""
def __init__(self):
"""C-tor."""
self.conn = None
def connect(self):
"""Connect to device, otherwise raise NotFoundError."""
raise NotImplementedError()
def __enter__(self):
"""Allow usage as context manager."""
self.conn = self.connect()
return self
def __exit__(self, *args):
"""Close and mark as disconnected."""
try:
self.conn.close()
except Exception as e: # pylint: disable=broad-except
log.exception('close failed: %s', e)
self.conn = None
def pubkey(self, identity, ecdh=False):
"""Get public key (as bytes)."""
raise NotImplementedError()
def sign(self, identity, blob):
"""Sign given blob and return the signature (as bytes)."""
raise NotImplementedError()
def ecdh(self, identity, pubkey):
"""Get shared session key using Elliptic Curve Diffie-Hellman."""
raise NotImplementedError()
def __str__(self):
"""Human-readable representation."""
return '{}'.format(self.__class__.__name__)

View File

@@ -0,0 +1,37 @@
"""KeepKey-related code (see https://www.keepkey.com/)."""
from . import trezor
from .. import formats
def _verify_support(identity, ecdh):
"""Make sure the device supports given configuration."""
protocol = identity.identity_dict['proto']
if protocol not in {'ssh'}:
raise NotImplementedError(
'Unsupported protocol: {}'.format(protocol))
if ecdh:
raise NotImplementedError('No support for ECDH')
if identity.curve_name not in {formats.CURVE_NIST256}:
raise NotImplementedError(
'Unsupported elliptic curve: {}'.format(identity.curve_name))
class KeepKey(trezor.Trezor):
"""Connection to KeepKey device."""
@property
def _defs(self):
from . import keepkey_defs
return keepkey_defs
required_version = '>=1.0.4'
def pubkey(self, identity, ecdh=False):
"""Return public key."""
_verify_support(identity, ecdh)
return trezor.Trezor.pubkey(self, identity=identity, ecdh=ecdh)
def ecdh(self, identity, pubkey):
"""No support for ECDH in KeepKey firmware."""
_verify_support(identity, ecdh=True)

View File

@@ -0,0 +1,8 @@
"""KeepKey-related definitions."""
# pylint: disable=unused-import
from keepkeylib.client import KeepKeyClient as Client
from keepkeylib.client import CallException
from keepkeylib.transport_hid import HidTransport
from keepkeylib.messages_pb2 import PassphraseAck
from keepkeylib.types_pb2 import IdentityType

View File

@@ -0,0 +1,111 @@
"""Ledger-related code (see https://www.ledgerwallet.com/)."""
import binascii
import logging
import struct
from ledgerblue import comm
from . import interface
log = logging.getLogger(__name__)
def _expand_path(path):
"""Convert BIP32 path into bytes."""
return b''.join((struct.pack('>I', e) for e in path))
def _convert_public_key(ecdsa_curve_name, result):
"""Convert Ledger reply into PublicKey object."""
if ecdsa_curve_name == 'nist256p1':
if (result[64] & 1) != 0:
result = bytearray([0x03]) + result[1:33]
else:
result = bytearray([0x02]) + result[1:33]
else:
result = result[1:]
keyX = bytearray(result[0:32])
keyY = bytearray(result[32:][::-1])
if (keyX[31] & 1) != 0:
keyY[31] |= 0x80
result = b'\x00' + bytes(keyY)
return bytes(result)
class LedgerNanoS(interface.Device):
"""Connection to Ledger Nano S device."""
def connect(self):
"""Enumerate and connect to the first USB HID interface."""
try:
return comm.getDongle()
except comm.CommException as e:
raise interface.NotFoundError(
'{} not connected: "{}"'.format(self, e))
def pubkey(self, identity, ecdh=False):
"""Get PublicKey object for specified BIP32 address and elliptic curve."""
curve_name = identity.get_curve_name(ecdh)
path = _expand_path(identity.get_bip32_address(ecdh))
if curve_name == 'nist256p1':
p2 = '01'
else:
p2 = '02'
apdu = '800200' + p2
apdu = binascii.unhexlify(apdu)
apdu += bytearray([len(path) + 1, len(path) // 4])
apdu += path
result = bytearray(self.conn.exchange(bytes(apdu)))[1:]
return _convert_public_key(curve_name, result)
def sign(self, identity, blob):
"""Sign given blob and return the signature (as bytes)."""
path = _expand_path(identity.get_bip32_address(ecdh=False))
if identity.identity_dict['proto'] == 'ssh':
ins = '04'
p1 = '00'
else:
ins = '08'
p1 = '00'
if identity.curve_name == 'nist256p1':
p2 = '81' if identity.identity_dict['proto'] == 'ssh' else '01'
else:
p2 = '82' if identity.identity_dict['proto'] == 'ssh' else '02'
apdu = '80' + ins + p1 + p2
apdu = binascii.unhexlify(apdu)
apdu += bytearray([len(blob) + len(path) + 1])
apdu += bytearray([len(path) // 4]) + path
apdu += blob
result = bytearray(self.conn.exchange(bytes(apdu)))
if identity.curve_name == 'nist256p1':
offset = 3
length = result[offset]
r = result[offset+1:offset+1+length]
if r[0] == 0:
r = r[1:]
offset = offset + 1 + length + 1
length = result[offset]
s = result[offset+1:offset+1+length]
if s[0] == 0:
s = s[1:]
offset = offset + 1 + length
return bytes(r) + bytes(s)
else:
return bytes(result[:64])
def ecdh(self, identity, pubkey):
"""Get shared session key using Elliptic Curve Diffie-Hellman."""
path = _expand_path(identity.get_bip32_address(ecdh=True))
if identity.curve_name == 'nist256p1':
p2 = '01'
else:
p2 = '02'
apdu = '800a00' + p2
apdu = binascii.unhexlify(apdu)
apdu += bytearray([len(pubkey) + len(path) + 1])
apdu += bytearray([len(path) // 4]) + path
apdu += pubkey
result = bytearray(self.conn.exchange(bytes(apdu)))
assert result[0] == 0x04
return bytes(result)

View File

@@ -0,0 +1,108 @@
"""TREZOR-related code (see http://bitcointrezor.com/)."""
import binascii
import logging
import semver
from . import interface
log = logging.getLogger(__name__)
class Trezor(interface.Device):
"""Connection to TREZOR device."""
@property
def _defs(self):
from . import trezor_defs
return trezor_defs
required_version = '>=1.4.0'
def connect(self):
"""Enumerate and connect to the first USB HID interface."""
def empty_passphrase_handler(_):
return self._defs.PassphraseAck(passphrase='')
for d in self._defs.HidTransport.enumerate():
log.debug('endpoint: %s', d)
transport = self._defs.HidTransport(d)
connection = self._defs.Client(transport)
connection.callback_PassphraseRequest = empty_passphrase_handler
f = connection.features
log.debug('connected to %s %s', self, f.device_id)
log.debug('label : %s', f.label)
log.debug('vendor : %s', f.vendor)
current_version = '{}.{}.{}'.format(f.major_version,
f.minor_version,
f.patch_version)
log.debug('version : %s', current_version)
log.debug('revision : %s', binascii.hexlify(f.revision))
if not semver.match(current_version, self.required_version):
fmt = ('Please upgrade your {} firmware to {} version'
' (current: {})')
raise ValueError(fmt.format(self, self.required_version,
current_version))
connection.ping(msg='', pin_protection=True) # unlock PIN
return connection
raise interface.NotFoundError('{} not connected'.format(self))
def close(self):
"""Close connection."""
self.conn.close()
def pubkey(self, identity, ecdh=False):
"""Return public key."""
curve_name = identity.get_curve_name(ecdh=ecdh)
log.debug('"%s" getting public key (%s) from %s',
identity, curve_name, self)
addr = identity.get_bip32_address(ecdh=ecdh)
result = self.conn.get_public_node(n=addr,
ecdsa_curve_name=curve_name)
log.debug('result: %s', result)
return result.node.public_key
def _identity_proto(self, identity):
result = self._defs.IdentityType()
for name, value in identity.items():
setattr(result, name, value)
return result
def sign(self, identity, blob):
"""Sign given blob and return the signature (as bytes)."""
curve_name = identity.get_curve_name(ecdh=False)
log.debug('"%s" signing %r (%s) on %s',
identity, blob, curve_name, self)
try:
result = self.conn.sign_identity(
identity=self._identity_proto(identity),
challenge_hidden=blob,
challenge_visual='',
ecdsa_curve_name=curve_name)
log.debug('result: %s', result)
assert len(result.signature) == 65
assert result.signature[:1] == b'\x00'
return result.signature[1:]
except self._defs.CallException as e:
msg = '{} error: {}'.format(self, e)
log.debug(msg, exc_info=True)
raise interface.DeviceError(msg)
def ecdh(self, identity, pubkey):
"""Get shared session key using Elliptic Curve Diffie-Hellman."""
curve_name = identity.get_curve_name(ecdh=True)
log.debug('"%s" shared session key (%s) for %r from %s',
identity, curve_name, pubkey, self)
try:
result = self.conn.get_ecdh_session_key(
identity=self._identity_proto(identity),
peer_public_key=pubkey,
ecdsa_curve_name=curve_name)
log.debug('result: %s', result)
assert len(result.session_key) in {65, 33} # NIST256 or Curve25519
assert result.session_key[:1] == b'\x04'
return result.session_key
except self._defs.CallException as e:
msg = '{} error: {}'.format(self, e)
log.debug(msg, exc_info=True)
raise interface.DeviceError(msg)

View File

@@ -0,0 +1,8 @@
"""TREZOR-related definitions."""
# pylint: disable=unused-import
from trezorlib.client import TrezorClient as Client
from trezorlib.client import CallException
from trezorlib.transport_hid import HidTransport
from trezorlib.messages_pb2 import PassphraseAck
from trezorlib.types_pb2 import IdentityType

View File

@@ -1,99 +0,0 @@
"""Thin wrapper around trezor/keepkey libraries."""
import binascii
import collections
import logging
import semver
log = logging.getLogger(__name__)
ClientWrapper = collections.namedtuple(
'ClientWrapper',
['connection', 'identity_type', 'device_name', 'call_exception'])
# pylint: disable=too-many-arguments
def _load_client(name, client_type, hid_transport,
passphrase_ack, identity_type,
required_version, call_exception):
def empty_passphrase_handler(_):
return passphrase_ack(passphrase='')
for d in hid_transport.enumerate():
connection = client_type(hid_transport(d))
connection.callback_PassphraseRequest = empty_passphrase_handler
f = connection.features
log.debug('connected to %s %s', name, f.device_id)
log.debug('label : %s', f.label)
log.debug('vendor : %s', f.vendor)
current_version = '{}.{}.{}'.format(f.major_version,
f.minor_version,
f.patch_version)
log.debug('version : %s', current_version)
log.debug('revision : %s', binascii.hexlify(f.revision))
if not semver.match(current_version, required_version):
fmt = 'Please upgrade your {} firmware to {} version (current: {})'
raise ValueError(fmt.format(name,
required_version,
current_version))
yield ClientWrapper(connection=connection,
identity_type=identity_type,
device_name=name,
call_exception=call_exception)
def _load_trezor():
try:
from trezorlib.client import TrezorClient, CallException
from trezorlib.transport_hid import HidTransport
from trezorlib.messages_pb2 import PassphraseAck
from trezorlib.types_pb2 import IdentityType
return _load_client(name='Trezor',
client_type=TrezorClient,
hid_transport=HidTransport,
passphrase_ack=PassphraseAck,
identity_type=IdentityType,
required_version='>=1.4.0',
call_exception=CallException)
except ImportError:
log.exception('Missing module: install via "pip install trezor"')
def _load_keepkey():
try:
from keepkeylib.client import KeepKeyClient, CallException
from keepkeylib.transport_hid import HidTransport
from keepkeylib.messages_pb2 import PassphraseAck
from keepkeylib.types_pb2 import IdentityType
return _load_client(name='KeepKey',
client_type=KeepKeyClient,
hid_transport=HidTransport,
passphrase_ack=PassphraseAck,
identity_type=IdentityType,
required_version='>=1.0.4',
call_exception=CallException)
except ImportError:
log.exception('Missing module: install via "pip install keepkey"')
LOADERS = [
_load_trezor,
_load_keepkey
]
def load(loaders=None):
"""Load a single device, via specified loaders' list."""
loaders = loaders if loaders is not None else LOADERS
device_list = []
for loader in loaders:
device = loader()
if device:
device_list.extend(device)
if len(device_list) == 1:
return device_list[0]
msg = '{:d} devices found'.format(len(device_list))
raise IOError(msg)

View File

@@ -11,11 +11,15 @@ from . import util
log = logging.getLogger(__name__)
# Supported ECDSA curves
# Supported ECDSA curves (for SSH and GPG)
CURVE_NIST256 = 'nist256p1'
CURVE_ED25519 = 'ed25519'
SUPPORTED_CURVES = {CURVE_NIST256, CURVE_ED25519}
# Supported ECDH curves (for GPG)
ECDH_NIST256 = 'nist256p1'
ECDH_CURVE25519 = 'curve25519'
# SSH key types
SSH_NIST256_DER_OCTET = b'\x04'
SSH_NIST256_KEY_PREFIX = b'ecdsa-sha2-'
@@ -134,7 +138,8 @@ def decompress_pubkey(pubkey, curve_name):
if len(pubkey) == 33:
decompress = {
CURVE_NIST256: _decompress_nist256,
CURVE_ED25519: _decompress_ed25519
CURVE_ED25519: _decompress_ed25519,
ECDH_CURVE25519: _decompress_ed25519,
}[curve_name]
vk = decompress(pubkey)
@@ -192,3 +197,12 @@ def import_public_key(line):
assert result['type'] == file_type.encode('ascii')
log.debug('loaded %s public key: %s', file_type, result['fingerprint'])
return result
def get_ecdh_curve_name(signature_curve_name):
"""Return appropriate curve for ECDH for specified signing curve."""
return {
CURVE_NIST256: ECDH_NIST256,
CURVE_ED25519: ECDH_CURVE25519,
ECDH_CURVE25519: ECDH_CURVE25519,
}[signature_curve_name]

View File

@@ -7,93 +7,120 @@ import os
import sys
import time
from . import agent, encode, keyring, protocol
from .. import server
import semver
from . import agent, client, encode, keyring, protocol
from .. import device, formats, server, util
log = logging.getLogger(__name__)
def run_create(args):
"""Generate a new pubkey for a new/existing GPG identity."""
user_id = os.environ['TREZOR_GPG_USER_ID']
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)
conn = encode.HardwareSigner(user_id=user_id,
curve_name=args.ecdsa_curve)
verifying_key = conn.pubkey(ecdh=False)
decryption_key = conn.pubkey(ecdh=True)
d = client.Client(user_id=args.user_id, curve_name=args.ecdsa_curve)
verifying_key = d.pubkey(ecdh=False)
decryption_key = d.pubkey(ecdh=True)
if args.subkey:
primary_bytes = keyring.export_public_key(user_id=user_id)
if args.subkey: # add as subkey
log.info('adding %s GPG subkey for "%s" to existing key',
args.ecdsa_curve, args.user_id)
# subkey for signing
signing_key = protocol.PublicKey(
curve_name=args.ecdsa_curve, created=args.time,
verifying_key=verifying_key, ecdh=False)
# subkey for encryption
encryption_key = protocol.PublicKey(
curve_name=args.ecdsa_curve, created=args.time,
verifying_key=decryption_key, ecdh=True)
curve_name=formats.get_ecdh_curve_name(args.ecdsa_curve),
created=args.time, verifying_key=decryption_key, ecdh=True)
primary_bytes = keyring.export_public_key(args.user_id)
result = encode.create_subkey(primary_bytes=primary_bytes,
pubkey=signing_key,
signer_func=conn.sign)
subkey=signing_key,
signer_func=d.sign)
result = encode.create_subkey(primary_bytes=result,
pubkey=encryption_key,
signer_func=conn.sign)
else:
subkey=encryption_key,
signer_func=d.sign)
else: # add as primary
log.info('creating new %s GPG primary key for "%s"',
args.ecdsa_curve, args.user_id)
# primary key for signing
primary = protocol.PublicKey(
curve_name=args.ecdsa_curve, created=args.time,
verifying_key=verifying_key, ecdh=False)
# subkey for encryption
subkey = protocol.PublicKey(
curve_name=args.ecdsa_curve, created=args.time,
verifying_key=decryption_key, ecdh=True)
curve_name=formats.get_ecdh_curve_name(args.ecdsa_curve),
created=args.time, verifying_key=decryption_key, ecdh=True)
result = encode.create_primary(user_id=user_id,
result = encode.create_primary(user_id=args.user_id,
pubkey=primary,
signer_func=conn.sign)
signer_func=d.sign)
result = encode.create_subkey(primary_bytes=result,
pubkey=subkey,
signer_func=conn.sign)
subkey=subkey,
signer_func=d.sign)
sys.stdout.write(protocol.armor(result, 'PUBLIC KEY BLOCK'))
def run_agent(args): # pylint: disable=unused-argument
def main_create():
"""Main function for GPG identity creation."""
p = argparse.ArgumentParser()
p.add_argument('user_id')
p.add_argument('-e', '--ecdsa-curve', default='nist256p1')
p.add_argument('-t', '--time', type=int, default=int(time.time()))
p.add_argument('-v', '--verbose', default=0, action='count')
p.add_argument('-s', '--subkey', default=False, action='store_true')
args = p.parse_args()
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!')
existing_gpg = keyring.gpg_version().decode('ascii')
required_gpg = '>=2.1.11'
if semver.match(existing_gpg, required_gpg):
run_create(args)
else:
log.error('Existing gpg2 has version "%s" (%s required)',
existing_gpg, required_gpg)
def main_agent():
"""Run a simple GPG-agent server."""
home_dir = os.environ.get('GNUPGHOME', os.path.expanduser('~/.gnupg/trezor'))
config_file = os.path.join(home_dir, 'gpg-agent.conf')
if not os.path.exists(config_file):
msg = 'No configuration file found: {}'.format(config_file)
raise IOError(msg)
lines = (line.strip() for line in open(config_file))
lines = (line for line in lines if line and not line.startswith('#'))
config = dict(line.split(' ', 1) for line in lines)
util.setup_logging(verbosity=int(config['verbosity']),
filename=config['log-file'])
sock_path = keyring.get_agent_sock_path()
with server.unix_domain_socket_server(sock_path) as sock:
for conn in agent.yield_connections(sock):
with contextlib.closing(conn):
agent.handle_connection(conn)
try:
agent.handle_connection(conn)
except StopIteration:
log.info('stopping gpg-agent')
return
except Exception as e: # pylint: disable=broad-except
log.exception('gpg-agent failed: %s', e)
def main():
"""Main function."""
def auto_unlock():
"""Automatically unlock first found device (used for `gpg-shell`)."""
p = argparse.ArgumentParser()
p.add_argument('-v', '--verbose', action='store_true', default=False)
subparsers = p.add_subparsers()
subparsers.required = True
subparsers.dest = 'command'
create_cmd = subparsers.add_parser('create')
create_cmd.add_argument('-s', '--subkey', action='store_true', default=False)
create_cmd.add_argument('-e', '--ecdsa-curve', default='nist256p1')
create_cmd.add_argument('-t', '--time', type=int, default=int(time.time()))
create_cmd.set_defaults(run=run_create)
agent_cmd = subparsers.add_parser('agent')
agent_cmd.set_defaults(run=run_agent)
p.add_argument('-v', '--verbose', default=0, action='count')
args = p.parse_args()
logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO,
format='%(asctime)s %(levelname)-10s %(message)s')
log.warning('This GPG tool is still in EXPERIMENTAL mode, '
'so please note that the API and features may '
'change without backwards compatibility!')
args.run(args)
if __name__ == '__main__':
main()
util.setup_logging(verbosity=args.verbose)
d = device.detect()
log.info('unlocked %s device', d)

View File

@@ -1,10 +1,8 @@
"""GPG-agent utilities."""
import binascii
import contextlib
import logging
import os
from . import decode, encode, keyring
from . import decode, client, keyring, protocol
from .. import util
log = logging.getLogger(__name__)
@@ -25,8 +23,9 @@ def yield_connections(sock):
def serialize(data):
"""Serialize data according to ASSUAN protocol."""
for c in ['%', '\n', '\r']:
data = data.replace(c, '%{:02X}'.format(ord(c)))
for c in [b'%', b'\n', b'\r']:
escaped = '%{:02X}'.format(ord(c)).encode('ascii')
data = data.replace(c, escaped)
return data
@@ -34,37 +33,54 @@ def sig_encode(r, s):
"""Serialize ECDSA signature data into GPG S-expression."""
r = serialize(util.num2bytes(r, 32))
s = serialize(util.num2bytes(s, 32))
return '(7:sig-val(5:ecdsa(1:r32:{})(1:s32:{})))'.format(r, s)
return b'(7:sig-val(5:ecdsa(1:r32:' + r + b')(1:s32:' + s + b')))'
def open_connection(keygrip_bytes):
"""
Connect to the device for the specified keygrip.
Parse GPG public key to find the first user ID, which is used to
specify the correct signature/decryption key on the device.
"""
pubkey_dict, user_ids = decode.load_by_keygrip(
pubkey_bytes=keyring.export_public_keys(),
keygrip=keygrip_bytes)
# We assume the first user ID is used to generate TREZOR-based GPG keys.
user_id = user_ids[0]['value'].decode('ascii')
curve_name = protocol.get_curve_name_by_oid(pubkey_dict['curve_oid'])
ecdh = (pubkey_dict['algo'] == protocol.ECDH_ALGO_ID)
conn = client.Client(user_id, curve_name=curve_name)
pubkey = protocol.PublicKey(
curve_name=curve_name, created=pubkey_dict['created'],
verifying_key=conn.pubkey(ecdh=ecdh), ecdh=ecdh)
assert pubkey.key_id() == pubkey_dict['key_id']
assert pubkey.keygrip() == keygrip_bytes
return conn
def pksign(keygrip, digest, algo):
"""Sign a message digest using a private EC key."""
assert algo == '8'
user_id = os.environ['TREZOR_GPG_USER_ID']
pubkey_dict = decode.load_public_key(
pubkey_bytes=keyring.export_public_key(user_id=user_id),
use_custom=True, ecdh=False)
pubkey, conn = encode.load_from_public_key(pubkey_dict=pubkey_dict)
with contextlib.closing(conn):
assert pubkey.keygrip == binascii.unhexlify(keygrip)
r, s = conn.sign(binascii.unhexlify(digest))
result = sig_encode(r, s)
log.debug('result: %r', result)
return result
log.debug('signing %r digest (algo #%s)', digest, algo)
keygrip_bytes = binascii.unhexlify(keygrip)
conn = open_connection(keygrip_bytes)
r, s = conn.sign(binascii.unhexlify(digest))
result = sig_encode(r, s)
log.debug('result: %r', result)
return result
def _serialize_point(data):
data = '{}:'.format(len(data)) + data
prefix = '{}:'.format(len(data)).encode('ascii')
# https://www.gnupg.org/documentation/manuals/assuan/Server-responses.html
for c in ['%', '\n', '\r']:
data = data.replace(c, '%{:02X}'.format(ord(c)))
return '(5:value' + data + ')'
return b'(5:value' + serialize(prefix + data) + b')'
def parse_ecdh(line):
"""Parse ECDH request and return remote public key."""
prefix, line = line.split(' ', 1)
assert prefix == 'D'
prefix, line = line.split(b' ', 1)
assert prefix == b'D'
exp, leftover = keyring.parse(keyring.unescape(line))
log.debug('ECDH s-exp: %r', exp)
assert not leftover
@@ -73,7 +89,7 @@ def parse_ecdh(line):
assert exp[0] == b'ecdh'
items = exp[1:]
log.debug('ECDH parameters: %r', items)
return dict(items)['e']
return dict(items)[b'e']
def pkdecrypt(keygrip, conn):
@@ -85,18 +101,9 @@ def pkdecrypt(keygrip, conn):
assert keyring.recvline(conn) == b'END'
remote_pubkey = parse_ecdh(line)
user_id = os.environ['TREZOR_GPG_USER_ID']
local_pubkey = decode.load_public_key(
pubkey_bytes=keyring.export_public_key(user_id=user_id),
use_custom=True, ecdh=True)
pubkey, conn = encode.load_from_public_key(pubkey_dict=local_pubkey)
with contextlib.closing(conn):
assert pubkey.keygrip == binascii.unhexlify(keygrip)
shared_secret = conn.ecdh(remote_pubkey)
assert len(shared_secret) == 65
assert shared_secret[:1] == b'\x04'
return _serialize_point(shared_secret)
keygrip_bytes = binascii.unhexlify(keygrip)
conn = open_connection(keygrip_bytes)
return _serialize_point(conn.ecdh(remote_pubkey))
def handle_connection(conn):
@@ -104,31 +111,40 @@ def handle_connection(conn):
keygrip = None
digest = None
algo = None
version = keyring.gpg_version()
version = keyring.gpg_version() # "Clone" existing GPG version
keyring.sendline(conn, b'OK')
for line in keyring.iterlines(conn):
parts = line.split(' ')
parts = line.split(b' ')
command = parts[0]
args = parts[1:]
if command in {'RESET', 'OPTION', 'HAVEKEY', 'SETKEYDESC'}:
if command in {b'RESET', b'OPTION', b'HAVEKEY', b'SETKEYDESC'}:
pass # reply with OK
elif command == 'GETINFO':
elif command == b'GETINFO':
keyring.sendline(conn, b'D ' + version)
elif command == 'AGENT_ID':
keyring.sendline(conn, b'D TREZOR')
elif command in {'SIGKEY', 'SETKEY'}:
elif command == b'AGENT_ID':
keyring.sendline(conn, b'D TREZOR') # "Fake" agent ID
elif command in {b'SIGKEY', b'SETKEY'}:
keygrip, = args
elif command == 'SETHASH':
elif command == b'SETHASH':
algo, digest = args
elif command == 'PKSIGN':
elif command == b'PKSIGN':
sig = pksign(keygrip, digest, algo)
keyring.sendline(conn, b'D ' + sig)
elif command == 'PKDECRYPT':
elif command == b'PKDECRYPT':
sec = pkdecrypt(keygrip, conn)
keyring.sendline(conn, b'D ' + sec)
elif command == 'BYE':
elif command == b'KEYINFO':
keygrip, = args
# Dummy reply (mainly for 'gpg --edit' to succeed).
# For details, see GnuPG agent KEYINFO command help.
fmt = 'S KEYINFO {0} X - - - - - - -'
keyring.sendline(conn, fmt.format(keygrip).encode('ascii'))
elif command == b'BYE':
return
elif command == b'KILLAGENT':
keyring.sendline(conn, b'OK')
raise StopIteration
else:
log.error('unknown request: %r', line)
return

View File

@@ -0,0 +1,44 @@
"""Device abstraction layer for GPG operations."""
import logging
from .. import device, formats, util
log = logging.getLogger(__name__)
class Client(object):
"""Sign messages and get public keys from a hardware device."""
def __init__(self, user_id, curve_name):
"""Connect to the device and retrieve required public key."""
self.device = device.detect()
self.user_id = user_id
self.identity = device.interface.Identity(
identity_str='gpg://', curve_name=curve_name)
self.identity.identity_dict['host'] = user_id
def pubkey(self, ecdh=False):
"""Return public key as VerifyingKey object."""
with self.device:
pubkey = self.device.pubkey(ecdh=ecdh, identity=self.identity)
return formats.decompress_pubkey(
pubkey=pubkey, curve_name=self.identity.curve_name)
def sign(self, digest):
"""Sign the digest and return a serialized signature."""
log.info('please confirm GPG signature on %s for "%s"...',
self.device, self.user_id)
if self.identity.curve_name == formats.CURVE_NIST256:
digest = digest[:32] # sign the first 256 bits
log.debug('signing digest: %s', util.hexlify(digest))
with self.device:
sig = self.device.sign(blob=digest, identity=self.identity)
return (util.bytes2num(sig[:32]), util.bytes2num(sig[32:]))
def ecdh(self, pubkey):
"""Derive shared secret using ECDH from remote public key."""
log.info('please confirm GPG decryption on %s for "%s"...',
self.device, self.user_id)
with self.device:
return self.device.ecdh(pubkey=pubkey, identity=self.identity)

View File

@@ -1,6 +1,5 @@
"""Decoders for GPG v2 data structures."""
import base64
import copy
import functools
import hashlib
import io
@@ -53,39 +52,29 @@ def parse_mpis(s, n):
return [parse_mpi(s) for _ in range(n)]
def _parse_nist256p1_verifier(mpi):
def _parse_nist256p1_pubkey(mpi):
prefix, x, y = util.split_bits(mpi, 4, 256, 256)
assert prefix == 4
point = ecdsa.ellipticcurve.Point(curve=ecdsa.NIST256p.curve,
x=x, y=y)
vk = ecdsa.VerifyingKey.from_public_point(
return ecdsa.VerifyingKey.from_public_point(
point=point, curve=ecdsa.curves.NIST256p,
hashfunc=hashlib.sha256)
def _nist256p1_verify(signature, digest):
result = vk.verify_digest(signature=signature,
digest=digest,
sigdecode=lambda rs, order: rs)
log.debug('nist256p1 ECDSA signature is OK (%s)', result)
return _nist256p1_verify, vk
def _parse_ed25519_verifier(mpi):
def _parse_ed25519_pubkey(mpi):
prefix, value = util.split_bits(mpi, 8, 256)
assert prefix == 0x40
vk = ed25519.VerifyingKey(util.num2bytes(value, size=32))
def _ed25519_verify(signature, digest):
sig = b''.join(util.num2bytes(val, size=32)
for val in signature)
result = vk.verify(sig, digest)
log.debug('ed25519 ECDSA signature is OK (%s)', result)
return _ed25519_verify, vk
return ed25519.VerifyingKey(util.num2bytes(value, size=32))
SUPPORTED_CURVES = {
b'\x2A\x86\x48\xCE\x3D\x03\x01\x07': _parse_nist256p1_verifier,
b'\x2B\x06\x01\x04\x01\xDA\x47\x0F\x01': _parse_ed25519_verifier,
b'\x2A\x86\x48\xCE\x3D\x03\x01\x07':
(_parse_nist256p1_pubkey, protocol.keygrip_nist256),
b'\x2B\x06\x01\x04\x01\xDA\x47\x0F\x01':
(_parse_ed25519_pubkey, protocol.keygrip_ed25519),
b'\x2B\x06\x01\x04\x01\x97\x55\x01\x05\x01':
(_parse_ed25519_pubkey, protocol.keygrip_curve25519),
}
RSA_ALGO_IDS = {1, 2, 3}
@@ -94,18 +83,6 @@ DSA_ALGO_ID = 17
ECDSA_ALGO_IDS = {18, 19, 22} # {ecdsa, nist256, ed25519}
def _parse_literal(stream):
"""See https://tools.ietf.org/html/rfc4880#section-5.9 for details."""
p = {'type': 'literal'}
p['format'] = stream.readfmt('c')
filename_len = stream.readfmt('B')
p['filename'] = stream.read(filename_len)
p['date'] = stream.readfmt('>L')
p['content'] = stream.read()
p['_to_hash'] = p['content']
return p
def _parse_embedded_signatures(subpackets):
for packet in subpackets:
data = bytearray(packet)
@@ -115,6 +92,12 @@ def _parse_embedded_signatures(subpackets):
yield _parse_signature(util.Reader(stream))
def has_custom_subpacket(signature_packet):
"""Detect our custom public keys by matching subpacket data."""
return any(protocol.CUSTOM_KEY_LABEL == subpacket[1:]
for subpacket in signature_packet['unhashed_subpackets'])
def _parse_signature(stream):
"""See https://tools.ietf.org/html/rfc4880#section-5.2 for details."""
p = {'type': 'signature'}
@@ -138,8 +121,6 @@ def _parse_signature(stream):
log.debug('embedded sigs: %s', embedded)
p['embedded'] = embedded
p['_is_custom'] = (protocol.CUSTOM_SUBPACKET in p['unhashed_subpackets'])
p['hash_prefix'] = stream.readfmt('2s')
if p['pubkey_alg'] in ECDSA_ALGO_IDS:
p['sig'] = (parse_mpi(stream), parse_mpi(stream))
@@ -168,11 +149,10 @@ def _parse_pubkey(stream, packet_type='pubkey'):
oid_size = stream.readfmt('B')
oid = stream.read(oid_size)
assert oid in SUPPORTED_CURVES, util.hexlify(oid)
parser = SUPPORTED_CURVES[oid]
p['curve_oid'] = oid
mpi = parse_mpi(stream)
log.debug('mpi: %x (%d bits)', mpi, mpi.bit_length())
p['verifier'], p['verifying_key'] = parser(mpi)
leftover = stream.read()
if leftover:
leftover = io.BytesIO(leftover)
@@ -180,16 +160,19 @@ def _parse_pubkey(stream, packet_type='pubkey'):
# should be b'\x03\x01\x08\x07': SHA256 + AES128
size, = util.readfmt(leftover, 'B')
p['kdf'] = leftover.read(size)
assert not leftover.read()
p['secret'] = leftover.read()
parse_func, keygrip_func = SUPPORTED_CURVES[oid]
keygrip = keygrip_func(parse_func(mpi))
log.debug('keygrip: %s', util.hexlify(keygrip))
p['keygrip'] = keygrip
elif p['algo'] == DSA_ALGO_ID:
log.warning('DSA signatures are not verified')
parse_mpis(stream, n=4)
parse_mpis(stream, n=4) # DSA keys are not supported
elif p['algo'] == ELGAMAL_ALGO_ID:
log.warning('ElGamal signatures are not verified')
parse_mpis(stream, n=3)
parse_mpis(stream, n=3) # ElGamal keys are not supported
else: # assume RSA
log.warning('RSA signatures are not verified')
parse_mpis(stream, n=2)
parse_mpis(stream, n=2) # RSA keys are not supported
assert not stream.read()
# https://tools.ietf.org/html/rfc4880#section-12.2
@@ -216,8 +199,9 @@ _parse_attribute = functools.partial(_parse_user_id,
PACKET_TYPES = {
2: _parse_signature,
5: _parse_pubkey,
6: _parse_pubkey,
11: _parse_literal,
7: _parse_subkey,
13: _parse_user_id,
14: _parse_subkey,
17: _parse_attribute,
@@ -280,28 +264,6 @@ def digest_packets(packets, hasher):
return hasher.digest()
def collect_packets(packets, types_to_collect):
"""Collect specified packet types into their leading packet."""
packet = None
result = []
for p in packets:
if p['type'] in types_to_collect:
packet.setdefault(p['type'], []).append(p)
else:
packet = copy.deepcopy(p)
result.append(packet)
return result
def parse_public_keys(stream):
"""Parse GPG public key into hierarchy of packets."""
packets = list(parse_packets(stream))
packets = collect_packets(packets, {'signature'})
packets = collect_packets(packets, {'user_id', 'user_attribute'})
packets = collect_packets(packets, {'subkey'})
return packets
HASH_ALGORITHMS = {
1: 'md5',
2: 'sha1',
@@ -313,42 +275,22 @@ HASH_ALGORITHMS = {
}
def load_public_key(pubkey_bytes, use_custom=False, ecdh=False):
"""Parse and validate GPG public key from an input stream."""
def load_by_keygrip(pubkey_bytes, keygrip):
"""Return public key and first user ID for specified keygrip."""
stream = io.BytesIO(pubkey_bytes)
packets = list(parse_packets(stream))
pubkey, userid, signature = packets[:3]
packets = packets[3:]
packets_per_pubkey = []
for p in packets:
if p['type'] == 'pubkey':
# Add a new packet list for each pubkey.
packets_per_pubkey.append([])
packets_per_pubkey[-1].append(p)
hash_alg = HASH_ALGORITHMS.get(signature['hash_alg'])
if hash_alg is not None:
digest = digest_packets(packets=[pubkey, userid, signature],
hasher=hashlib.new(hash_alg))
assert signature['hash_prefix'] == digest[:2]
log.debug('loaded public key "%s"', userid['value'])
if hash_alg is not None and pubkey.get('verifier'):
verify_digest(pubkey=pubkey, digest=digest,
signature=signature['sig'], label='GPG public key')
else:
log.warning('public key %s is not verified!',
util.hexlify(pubkey['key_id']))
packet = pubkey
while use_custom:
if packet['type'] in ('pubkey', 'subkey') and signature['_is_custom']:
if ecdh == (packet['algo'] == protocol.ECDH_ALGO_ID):
log.debug('found custom %s', packet['type'])
break
while packets[1]['type'] != 'signature':
packets = packets[1:]
packet, signature = packets[:2]
packets = packets[2:]
packet['user_id'] = userid['value']
packet['_is_custom'] = signature['_is_custom']
return packet
for packets in packets_per_pubkey:
user_ids = [p for p in packets if p['type'] == 'user_id']
for p in packets:
if p.get('keygrip') == keygrip:
return p, user_ids
def load_signature(stream, original_data):
@@ -361,17 +303,6 @@ def load_signature(stream, original_data):
return signature, digest
def verify_digest(pubkey, digest, signature, label):
"""Verify a digest signature from a specified public key."""
verifier = pubkey['verifier']
try:
verifier(signature, digest)
log.debug('%s is OK', label)
except ecdsa.keys.BadSignatureError:
log.error('Bad %s!', label)
raise ValueError('Invalid ECDSA signature for {}'.format(label))
def remove_armor(armored_data):
"""Decode armored data into its binary form."""
stream = io.BytesIO(armored_data)
@@ -380,11 +311,3 @@ def remove_armor(armored_data):
payload, checksum = data[:-3], data[-3:]
assert util.crc24(payload) == checksum
return payload
def verify(pubkey, signature, original_data):
"""Verify correctness of public key and signature."""
stream = io.BytesIO(remove_armor(signature))
signature, digest = load_signature(stream, original_data)
verify_digest(pubkey=pubkey, digest=digest,
signature=signature['sig'], label='GPG signature')

View File

@@ -1,93 +1,23 @@
"""Create GPG ECDSA signatures and public keys using TREZOR device."""
import io
import logging
import time
from . import decode, keyring, protocol
from .. import client, factory, formats, util
from .. import util
log = logging.getLogger(__name__)
class HardwareSigner(object):
"""Sign messages and get public keys from a hardware device."""
def __init__(self, user_id, curve_name):
"""Connect to the device and retrieve required public key."""
self.client_wrapper = factory.load()
self.identity = self.client_wrapper.identity_type()
self.identity.proto = 'gpg'
self.identity.host = user_id
self.curve_name = curve_name
def pubkey(self, ecdh=False):
"""Return public key as VerifyingKey object."""
addr = client.get_address(identity=self.identity, ecdh=ecdh)
public_node = self.client_wrapper.connection.get_public_node(
n=addr, ecdsa_curve_name=self.curve_name)
return formats.decompress_pubkey(
pubkey=public_node.node.public_key,
curve_name=self.curve_name)
def sign(self, digest):
"""Sign the digest and return a serialized signature."""
result = self.client_wrapper.connection.sign_identity(
identity=self.identity,
challenge_hidden=digest,
challenge_visual='',
ecdsa_curve_name=self.curve_name)
assert result.signature[:1] == b'\x00'
sig = result.signature[1:]
return (util.bytes2num(sig[:32]), util.bytes2num(sig[32:]))
def ecdh(self, pubkey):
"""Derive shared secret using ECDH from remote public key."""
result = self.client_wrapper.connection.get_ecdh_session_key(
identity=self.identity,
peer_public_key=pubkey,
ecdsa_curve_name=self.curve_name)
assert len(result.session_key) == 65
assert result.session_key[:1] == b'\x04'
return result.session_key
def close(self):
"""Close the connection to the device."""
self.client_wrapper.connection.clear_session()
self.client_wrapper.connection.close()
class AgentSigner(object):
"""Sign messages and get public keys using gpg-agent tool."""
def __init__(self, user_id):
"""Connect to the agent and retrieve required public key."""
self.sock = keyring.connect_to_agent()
self.keygrip = keyring.get_keygrip(user_id)
def sign(self, digest):
"""Sign the digest and return an ECDSA/RSA/DSA signature."""
return keyring.sign_digest(sock=self.sock,
keygrip=self.keygrip, digest=digest)
def close(self):
"""Close the connection to gpg-agent."""
self.sock.close()
def _time_format(t):
return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(t))
def create_primary(user_id, pubkey, signer_func):
def create_primary(user_id, pubkey, signer_func, secret_bytes=b''):
"""Export new primary GPG public key, ready for "gpg2 --import"."""
pubkey_packet = protocol.packet(tag=6, blob=pubkey.data())
pubkey_packet = protocol.packet(tag=(5 if secret_bytes else 6),
blob=(pubkey.data() + secret_bytes))
user_id_packet = protocol.packet(tag=13,
blob=user_id.encode('ascii'))
data_to_sign = (pubkey.data_to_hash() +
user_id_packet[:1] +
util.prefix_len('>L', user_id.encode('ascii')))
log.info('creating primary GPG key "%s"', user_id)
hashed_subpackets = [
protocol.subpacket_time(pubkey.created), # signature time
# https://tools.ietf.org/html/rfc4880#section-5.2.3.7
@@ -106,7 +36,6 @@ def create_primary(user_id, pubkey, signer_func):
protocol.subpacket(16, pubkey.key_id()), # issuer key id
protocol.CUSTOM_SUBPACKET]
log.info('confirm signing with primary key')
signature = protocol.make_signature(
signer_func=signer_func,
public_algo=pubkey.algo_id,
@@ -119,26 +48,27 @@ def create_primary(user_id, pubkey, signer_func):
return pubkey_packet + user_id_packet + sign_packet
def create_subkey(primary_bytes, pubkey, signer_func):
def create_subkey(primary_bytes, subkey, signer_func, secret_bytes=b''):
"""Export new subkey to GPG primary key."""
subkey_packet = protocol.packet(tag=14, blob=pubkey.data())
primary = decode.load_public_key(primary_bytes)
log.info('adding subkey to primary GPG key "%s"', primary['user_id'])
data_to_sign = primary['_to_hash'] + pubkey.data_to_hash()
subkey_packet = protocol.packet(tag=(7 if secret_bytes else 14),
blob=(subkey.data() + secret_bytes))
packets = list(decode.parse_packets(io.BytesIO(primary_bytes)))
primary, user_id, signature = packets[:3]
if pubkey.ecdh:
data_to_sign = primary['_to_hash'] + subkey.data_to_hash()
if subkey.ecdh:
embedded_sig = None
else:
# Primary Key Binding Signature
hashed_subpackets = [
protocol.subpacket_time(pubkey.created)] # signature time
protocol.subpacket_time(subkey.created)] # signature time
unhashed_subpackets = [
protocol.subpacket(16, pubkey.key_id())] # issuer key id
log.info('confirm signing with new subkey')
protocol.subpacket(16, subkey.key_id())] # issuer key id
embedded_sig = protocol.make_signature(
signer_func=signer_func,
data_to_sign=data_to_sign,
public_algo=pubkey.algo_id,
public_algo=subkey.algo_id,
sig_type=0x19,
hashed_subpackets=hashed_subpackets,
unhashed_subpackets=unhashed_subpackets)
@@ -147,10 +77,10 @@ def create_subkey(primary_bytes, pubkey, signer_func):
# Key flags: https://tools.ietf.org/html/rfc4880#section-5.2.3.21
# (certify & sign) (encrypt)
flags = (2) if (not pubkey.ecdh) else (4 | 8)
flags = (2) if (not subkey.ecdh) else (4 | 8)
hashed_subpackets = [
protocol.subpacket_time(pubkey.created), # signature time
protocol.subpacket_time(subkey.created), # signature time
protocol.subpacket_byte(0x1B, flags)]
unhashed_subpackets = []
@@ -159,9 +89,8 @@ def create_subkey(primary_bytes, pubkey, signer_func):
unhashed_subpackets.append(protocol.subpacket(32, embedded_sig))
unhashed_subpackets.append(protocol.CUSTOM_SUBPACKET)
log.info('confirm signing with primary key')
if not primary['_is_custom']:
signer_func = AgentSigner(primary['user_id']).sign
if not decode.has_custom_subpacket(signature):
signer_func = keyring.create_agent_signer(user_id['value'])
signature = protocol.make_signature(
signer_func=signer_func,
@@ -172,22 +101,3 @@ def create_subkey(primary_bytes, pubkey, signer_func):
unhashed_subpackets=unhashed_subpackets)
sign_packet = protocol.packet(tag=2, blob=signature)
return primary_bytes + subkey_packet + sign_packet
def load_from_public_key(pubkey_dict):
"""Load correct public key from the device."""
user_id = pubkey_dict['user_id']
created = pubkey_dict['created']
curve_name = protocol.find_curve_by_algo_id(pubkey_dict['algo'])
assert curve_name in formats.SUPPORTED_CURVES
ecdh = (pubkey_dict['algo'] == protocol.ECDH_ALGO_ID)
conn = HardwareSigner(user_id, curve_name=curve_name)
pubkey = protocol.PublicKey(
curve_name=curve_name, created=created,
verifying_key=conn.pubkey(ecdh=ecdh), ecdh=ecdh)
assert pubkey.key_id() == pubkey_dict['key_id']
log.info('%s created at %s for "%s"',
pubkey, _time_format(pubkey.created), user_id)
return pubkey, conn

View File

@@ -1,4 +1,5 @@
"""Tools for doing signature using gpg-agent."""
from __future__ import absolute_import, print_function, unicode_literals
import binascii
import io
@@ -15,15 +16,15 @@ log = logging.getLogger(__name__)
def get_agent_sock_path(sp=subprocess):
"""Parse gpgconf output to find out GPG agent UNIX socket path."""
lines = sp.check_output(['gpgconf', '--list-dirs']).strip().split('\n')
dirs = dict(line.split(':', 1) for line in lines)
return dirs['agent-socket']
lines = sp.check_output(['gpgconf', '--list-dirs']).strip().split(b'\n')
dirs = dict(line.split(b':', 1) for line in lines)
return dirs[b'agent-socket']
def connect_to_agent(sp=subprocess):
"""Connect to GPG agent's UNIX socket."""
sock_path = get_agent_sock_path(sp=sp)
sp.check_call(['gpg-connect-agent', '/bye'])
sp.check_call(['gpg-connect-agent', '/bye']) # Make sure it's running
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.connect(sock_path)
return sock
@@ -157,7 +158,12 @@ def sign_digest(sock, keygrip, digest, sp=subprocess, environ=None):
assert communicate(sock, 'SETKEYDESC '
'Sign+a+new+TREZOR-based+subkey') == b'OK'
assert communicate(sock, 'PKSIGN') == b'OK'
line = recvline(sock).strip()
while True:
line = recvline(sock).strip()
if line.startswith(b'S PROGRESS'):
continue
else:
break
line = unescape(line)
log.debug('unescaped: %r', line)
prefix, sig = line.split(b' ', 1)
@@ -190,7 +196,7 @@ def get_keygrip(user_id, sp=subprocess):
def gpg_version(sp=subprocess):
"""Get a keygrip of the primary GPG key of the specified user."""
args = gpg_command(['--version'])
output = sp.check_output(args).decode('ascii')
output = sp.check_output(args)
line = output.split(b'\n')[0] # b'gpg (GnuPG) 2.1.11'
return line.split(b' ')[-1] # b'2.1.11'
@@ -203,3 +209,21 @@ def export_public_key(user_id, sp=subprocess):
log.error('could not find public key %r in local GPG keyring', user_id)
raise KeyError(user_id)
return result
def export_public_keys(sp=subprocess):
"""Export all GPG public keys."""
args = gpg_command(['--export'])
return sp.check_output(args=args)
def create_agent_signer(user_id):
"""Sign digest with existing GPG keys using gpg-agent tool."""
sock = connect_to_agent()
keygrip = get_keygrip(user_id)
def sign(digest):
"""Sign the digest and return an ECDSA/RSA/DSA signature."""
return sign_digest(sock=sock, keygrip=keygrip, digest=digest)
return sign

View File

@@ -47,9 +47,22 @@ def subpacket_byte(subpacket_type, value):
return subpacket(subpacket_type, '>B', value)
def subpacket_prefix_len(item):
"""Prefix subpacket length according to RFC 4880 section-5.2.3.1."""
n = len(item)
if n >= 8384:
prefix = b'\xFF' + struct.pack('>L', n)
elif n >= 192:
n = n - 192
prefix = struct.pack('BB', (n // 256) + 192, n % 256)
else:
prefix = struct.pack('B', n)
return prefix + item
def subpackets(*items):
"""Serialize several GPG subpackets."""
prefixed = [util.prefix_len('>B', item) for item in items]
prefixed = [subpacket_prefix_len(item) for item in items]
return util.prefix_len('>H', b''.join(prefixed))
@@ -86,7 +99,8 @@ def _compute_keygrip(params):
return hashlib.sha1(b''.join(parts)).digest()
def _keygrip_nist256(vk):
def keygrip_nist256(vk):
"""Compute keygrip for NIST256 curve public keys."""
curve = vk.curve.curve
gen = vk.curve.generator
g = (4 << 512) | (gen.x() << 256) | gen.y()
@@ -103,7 +117,8 @@ def _keygrip_nist256(vk):
])
def _keygrip_ed25519(vk):
def keygrip_ed25519(vk):
"""Compute keygrip for Ed25519 public keys."""
# pylint: disable=line-too-long
return _compute_keygrip([
['p', util.num2bytes(0x7FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFED, size=32)], # nopep8
@@ -115,35 +130,54 @@ def _keygrip_ed25519(vk):
])
def keygrip_curve25519(vk):
"""Compute keygrip for Curve25519 public keys."""
# pylint: disable=line-too-long
return _compute_keygrip([
['p', util.num2bytes(0x7FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFED, size=32)], # nopep8
['a', b'\x01\xDB\x41'],
['b', b'\x01'],
['g', util.num2bytes(0x04000000000000000000000000000000000000000000000000000000000000000920ae19a1b8a086b4e01edd2c7748d14c923d4d7e6d7c61b229e9c5a27eced3d9, size=65)], # nopep8
['n', util.num2bytes(0x1000000000000000000000000000000014DEF9DEA2F79CD65812631A5CF5D3ED, size=32)], # nopep8
['q', vk.to_bytes()],
])
SUPPORTED_CURVES = {
formats.CURVE_NIST256: {
# https://tools.ietf.org/html/rfc6637#section-11
'oid': b'\x2A\x86\x48\xCE\x3D\x03\x01\x07',
'algo_id': 19,
'serialize': _serialize_nist256,
'keygrip': _keygrip_nist256,
'keygrip': keygrip_nist256,
},
formats.CURVE_ED25519: {
'oid': b'\x2B\x06\x01\x04\x01\xDA\x47\x0F\x01',
'algo_id': 22,
'serialize': _serialize_ed25519,
'keygrip': _keygrip_ed25519,
}
'keygrip': keygrip_ed25519,
},
formats.ECDH_CURVE25519: {
'oid': b'\x2B\x06\x01\x04\x01\x97\x55\x01\x05\x01',
'algo_id': 18,
'serialize': _serialize_ed25519,
'keygrip': keygrip_curve25519,
},
}
ECDH_ALGO_ID = 18
CUSTOM_SUBPACKET = subpacket(100, b'TREZOR-GPG') # marks "our" pubkey
CUSTOM_KEY_LABEL = b'TREZOR-GPG' # marks "our" pubkey
CUSTOM_SUBPACKET_ID = 26 # use "policy URL" subpacket
CUSTOM_SUBPACKET = subpacket(CUSTOM_SUBPACKET_ID, CUSTOM_KEY_LABEL)
def find_curve_by_algo_id(algo_id):
"""Find curve name that matches a public key algorith ID."""
if algo_id == ECDH_ALGO_ID:
return formats.CURVE_NIST256
curve_name, = [name for name, info in SUPPORTED_CURVES.items()
if info['algo_id'] == algo_id]
return curve_name
def get_curve_name_by_oid(oid):
"""Return curve name matching specified OID, or raise KeyError."""
for curve_name, info in SUPPORTED_CURVES.items():
if info['oid'] == oid:
return curve_name
raise KeyError('Unknown OID: {!r}'.format(oid))
class PublicKey(object):
@@ -151,10 +185,11 @@ class PublicKey(object):
def __init__(self, curve_name, created, verifying_key, ecdh=False):
"""Contruct using a ECDSA VerifyingKey object."""
self.curve_name = curve_name
self.curve_info = SUPPORTED_CURVES[curve_name]
self.created = int(created) # time since Epoch
self.verifying_key = verifying_key
self.ecdh = ecdh
self.ecdh = bool(ecdh)
if ecdh:
self.algo_id = ECDH_ALGO_ID
self.ecdh_packet = b'\x03\x01\x08\x07'
@@ -162,12 +197,8 @@ class PublicKey(object):
self.algo_id = self.curve_info['algo_id']
self.ecdh_packet = b''
hex_key_id = util.hexlify(self.key_id())[-8:]
self.desc = 'GPG public key {}/{}'.format(curve_name, hex_key_id)
@property
def keygrip(self):
"""Compute GPG2 keygrip."""
"""Compute GPG keygrip of the verifying key."""
return self.curve_info['keygrip'](self.verifying_key)
def data(self):
@@ -193,7 +224,8 @@ class PublicKey(object):
def __repr__(self):
"""Short (8 hexadecimal digits) GPG key ID."""
return self.desc
hex_key_id = util.hexlify(self.key_id())[-8:]
return 'GPG public key {}/{}'.format(self.curve_name, hex_key_id)
__str__ = __repr__

View File

@@ -1,11 +1,10 @@
import glob
import hashlib
import io
import os
import pytest
from .. import decode
from .. import decode, protocol
from ... import util
@@ -14,6 +13,15 @@ def test_subpackets():
assert decode.parse_subpackets(util.Reader(s)) == [b'\xAB\xCD', b'\xEF']
def test_subpackets_prefix():
for n in [0, 1, 2, 4, 5, 10, 191, 192, 193,
255, 256, 257, 8383, 8384, 65530]:
item = b'?' * n # create dummy subpacket
prefixed = protocol.subpackets(item)
result = decode.parse_subpackets(util.Reader(io.BytesIO(prefixed)))
assert [item] == result
def test_mpi():
s = io.BytesIO(b'\x00\x09\x01\x23')
assert decode.parse_mpi(util.Reader(s)) == 0x123
@@ -22,84 +30,6 @@ def test_mpi():
assert decode.parse_mpis(util.Reader(s), n=2) == [0x123, 5]
def assert_subdict(d, s):
for k, v in s.items():
assert d[k] == v
def test_primary_nist256p1():
# pylint: disable=line-too-long
data = b'''-----BEGIN PGP PUBLIC KEY BLOCK-----
Version: GnuPG v2
mFIEV0hI1hMIKoZIzj0DAQcCAwTXY7aq01xMPSU7gTHU9B7Z2CFoCk1Y4WYb8Tiy
hurvIZ5la6+UEgAKF9HXpQo0yE+HQOgufoLlCpdE7NoEUb+HtAd0ZXN0aW5niHYE
ExMIABIFAldISNYCGwMCFQgCFgACF4AAFgkQTcCehfpEIPILZFRSRVpPUi1HUEeV
3QEApHKmBkbLVZNpsB8q9mBzKytxnOHNB3QWDuoKJu/ERi4A/1wRGZ/B0BDazHck
zpR9luXTKwMEl+mlZmwEFKZXBmir
=oyj0
-----END PGP PUBLIC KEY BLOCK-----
'''
stream = io.BytesIO(decode.remove_armor(data))
pubkey, user_id, signature = list(decode.parse_packets(stream))
expected_pubkey = {
'created': 1464355030, 'type': 'pubkey', 'tag': 6,
'version': 4, 'algo': 19, 'key_id': b'M\xc0\x9e\x85\xfaD \xf2',
'_to_hash': b'\x99\x00R\x04WHH\xd6\x13\x08*\x86H\xce=\x03\x01\x07\x02\x03\x04\xd7c\xb6\xaa\xd3\\L=%;\x811\xd4\xf4\x1e\xd9\xd8!h\nMX\xe1f\x1b\xf18\xb2\x86\xea\xef!\x9eek\xaf\x94\x12\x00\n\x17\xd1\xd7\xa5\n4\xc8O\x87@\xe8.~\x82\xe5\n\x97D\xec\xda\x04Q\xbf\x87' # nopep8
}
assert_subdict(pubkey, expected_pubkey)
point = pubkey['verifying_key'].pubkey.point
assert point.x(), point.y() == (
97423441028100245505102979561460969898742433559010922791700160771755342491425,
71644624850142103522769833619875243486871666152651730678601507641225861250951
)
assert_subdict(user_id, {
'tag': 13, 'type': 'user_id', 'value': b'testing',
'_to_hash': b'\xb4\x00\x00\x00\x07testing'
})
assert_subdict(signature, {
'pubkey_alg': 19, '_is_custom': True, 'hash_alg': 8, 'tag': 2,
'sig_type': 19, 'version': 4, 'type': 'signature', 'hash_prefix': b'\x95\xdd',
'sig': (74381873592149178031432444136130575481350858387410643140628758456112511206958,
41642995320462795718437755373080464775445470754419831653624197847615308982443),
'hashed_subpackets': [b'\x02WHH\xd6', b'\x1b\x03', b'\x15\x08', b'\x16\x00', b'\x17\x80'],
'unhashed_subpackets': [b'\x10M\xc0\x9e\x85\xfaD \xf2', b'dTREZOR-GPG'],
'_to_hash': b'\x04\x13\x13\x08\x00\x12\x05\x02WHH\xd6\x02\x1b\x03\x02\x15\x08\x02\x16\x00\x02\x17\x80\x04\xff\x00\x00\x00\x18' # nopep8
})
digest = decode.digest_packets(packets=[pubkey, user_id, signature],
hasher=hashlib.sha256())
decode.verify_digest(pubkey=pubkey, digest=digest,
signature=signature['sig'],
label='GPG primary public key')
with pytest.raises(ValueError):
bad_digest = b'\x00' * len(digest)
decode.verify_digest(pubkey=pubkey, digest=bad_digest,
signature=signature['sig'],
label='GPG primary public key')
message = b'Hello, World!\n'
signature = b'''-----BEGIN PGP SIGNATURE-----
Version: GnuPG v2
iF4EABMIAAYFAldIlfQACgkQTcCehfpEIPKOUgD9FjaeWla4wOuDZ7P6fhkT5nZp
KDQU0N5KmNwLlt2kwo4A/jQkBII2cI8tTqOVTLNRXXqIOsMf/fG4jKM/VOFc/01c
=dC+z
-----END PGP SIGNATURE-----
'''
decode.verify(pubkey=pubkey, signature=signature, original_data=message)
pubkey = decode.load_public_key(pubkey_bytes=decode.remove_armor(data))
assert_subdict(pubkey, expected_pubkey)
assert_subdict(pubkey, {'user_id': b'testing'})
pubkey = decode.load_public_key(pubkey_bytes=decode.remove_armor(data),
use_custom=True)
assert_subdict(pubkey, expected_pubkey)
assert_subdict(pubkey, {'user_id': b'testing'})
cwd = os.path.join(os.path.dirname(__file__))
input_files = glob.glob(os.path.join(cwd, '*.gpg'))
@@ -111,5 +41,18 @@ def public_key_path(request):
def test_gpg_files(public_key_path): # pylint: disable=redefined-outer-name
with open(public_key_path, 'rb') as f:
keys = decode.parse_public_keys(f)
assert len(keys) > 0
packets = list(decode.parse_packets(f))
assert len(packets) > 0
def test_has_custom_subpacket():
sig = {'unhashed_subpackets': []}
assert not decode.has_custom_subpacket(sig)
custom_markers = [
protocol.CUSTOM_SUBPACKET,
protocol.subpacket(10, protocol.CUSTOM_KEY_LABEL),
]
for marker in custom_markers:
sig = {'unhashed_subpackets': [marker]}
assert decode.has_custom_subpacket(sig)

View File

@@ -1,4 +1,5 @@
import io
import mock
from .. import keyring
@@ -79,4 +80,22 @@ PKSIGN
def test_iterlines():
sock = FakeSocket()
sock.rx.write(b'foo\nbar\nxyz')
assert list(keyring.iterlines(sock)) == []
sock.rx.seek(0)
assert list(keyring.iterlines(sock)) == [b'foo', b'bar']
def test_get_agent_sock_path():
sp = mock.Mock(spec=['check_output'])
sp.check_output.return_value = b'''sysconfdir:/usr/local/etc/gnupg
bindir:/usr/local/bin
libexecdir:/usr/local/libexec
libdir:/usr/local/lib/gnupg
datadir:/usr/local/share/gnupg
localedir:/usr/local/share/locale
dirmngr-socket:/run/user/1000/gnupg/S.dirmngr
agent-ssh-socket:/run/user/1000/gnupg/S.gpg-agent.ssh
agent-socket:/run/user/1000/gnupg/S.gpg-agent
homedir:/home/roman/.gnupg
'''
expected = b'/run/user/1000/gnupg/S.gpg-agent'
assert keyring.get_agent_sock_path(sp=sp) == expected

View File

@@ -1,5 +1,6 @@
import ecdsa
import ed25519
import pytest
from .. import protocol
from ... import formats
@@ -30,11 +31,6 @@ def test_mpi():
assert protocol.mpi(0x123) == b'\x00\x09\x01\x23'
def test_find():
assert protocol.find_curve_by_algo_id(19) == formats.CURVE_NIST256
assert protocol.find_curve_by_algo_id(22) == formats.CURVE_ED25519
def test_armor():
data = bytearray(range(256))
assert protocol.armor(data, 'TEST') == '''-----BEGIN PGP TEST-----
@@ -74,7 +70,7 @@ def test_nist256p1():
pk = protocol.PublicKey(curve_name=formats.CURVE_NIST256,
created=42, verifying_key=vk)
assert repr(pk) == 'GPG public key nist256p1/F82361D9'
assert pk.keygrip == b'\x95\x85.\x91\x7f\xe2\xc3\x91R\xba\x99\x81\x92\xb5y\x1d\xb1\\\xdc\xf0'
assert pk.keygrip() == b'\x95\x85.\x91\x7f\xe2\xc3\x91R\xba\x99\x81\x92\xb5y\x1d\xb1\\\xdc\xf0'
def test_nist256p1_ecdh():
@@ -83,7 +79,7 @@ def test_nist256p1_ecdh():
pk = protocol.PublicKey(curve_name=formats.CURVE_NIST256,
created=42, verifying_key=vk, ecdh=True)
assert repr(pk) == 'GPG public key nist256p1/5811DF46'
assert pk.keygrip == b'\x95\x85.\x91\x7f\xe2\xc3\x91R\xba\x99\x81\x92\xb5y\x1d\xb1\\\xdc\xf0'
assert pk.keygrip() == b'\x95\x85.\x91\x7f\xe2\xc3\x91R\xba\x99\x81\x92\xb5y\x1d\xb1\\\xdc\xf0'
def test_ed25519():
@@ -92,4 +88,20 @@ def test_ed25519():
pk = protocol.PublicKey(curve_name=formats.CURVE_ED25519,
created=42, verifying_key=vk)
assert repr(pk) == 'GPG public key ed25519/36B40FE6'
assert pk.keygrip == b'\xbf\x01\x90l\x17\xb64\xa3-\xf4\xc0gr\x99\x18<\xddBQ?'
assert pk.keygrip() == b'\xbf\x01\x90l\x17\xb64\xa3-\xf4\xc0gr\x99\x18<\xddBQ?'
def test_curve25519():
sk = ed25519.SigningKey(b'\x00' * 32)
vk = sk.get_verifying_key()
pk = protocol.PublicKey(curve_name=formats.ECDH_CURVE25519,
created=42, verifying_key=vk)
assert repr(pk) == 'GPG public key curve25519/69460384'
assert pk.keygrip() == b'x\xd6\x86\xe4\xa6\xfc;\x0fY\xe1}Lw\xc4\x9ed\xf1Q\x8a\x00'
def test_get_curve_name_by_oid():
for name, info in protocol.SUPPORTED_CURVES.items():
assert protocol.get_curve_name_by_oid(info['oid']) == name
with pytest.raises(KeyError):
protocol.get_curve_name_by_oid('BAD_OID')

View File

@@ -7,7 +7,6 @@ for more details.
The server's source code can be found here:
https://github.com/openssh/openssh-portable/blob/master/authfd.c
"""
import binascii
import io
import logging
@@ -122,7 +121,7 @@ class Handler(object):
SSH v2 public key authentication is performed.
If the required key is not supported, raise KeyError
If the signature is invalid, rause ValueError
If the signature is invalid, raise ValueError
"""
key = formats.parse_pubkey(util.read_frame(buf))
log.debug('looking for %s', key['fingerprint'])
@@ -138,13 +137,13 @@ class Handler(object):
else:
raise KeyError('key not found')
log.debug('signing %d-byte blob', len(blob))
label = key['name'].decode('ascii') # label should be a string
log.debug('signing %d-byte blob with "%s" key', len(blob), label)
try:
signature = self.signer(label=label, blob=blob)
signature = self.signer(blob=blob, identity=key['identity'])
except IOError:
return failure()
log.debug('signature: %s', binascii.hexlify(signature))
log.debug('signature: %r', signature)
try:
sig_bytes = key['verifier'](sig=signature, msg=blob)

View File

@@ -1,5 +1,6 @@
"""UNIX-domain socket server for ssh-agent implementation."""
import contextlib
import functools
import logging
import os
import socket
@@ -30,7 +31,7 @@ def unix_domain_socket_server(sock_path):
Listen on it, and delete it after the generated context is over.
"""
log.debug('serving on SSH_AUTH_SOCK=%s', sock_path)
log.debug('serving on %s', sock_path)
remove_file(sock_path)
server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
@@ -42,19 +43,24 @@ def unix_domain_socket_server(sock_path):
remove_file(sock_path)
def handle_connection(conn, handler):
def handle_connection(conn, handler, mutex):
"""
Handle a single connection using the specified protocol handler in a loop.
Since this function may be called concurrently from server_thread,
the specified mutex is used to synchronize the device handling.
Exit when EOFError is raised.
All other exceptions are logged as warnings.
"""
try:
log.debug('welcome agent')
while True:
msg = util.read_frame(conn)
reply = handler.handle(msg=msg)
util.send(conn, reply)
with contextlib.closing(conn):
while True:
msg = util.read_frame(conn)
with mutex:
reply = handler.handle(msg=msg)
util.send(conn, reply)
except EOFError:
log.debug('goodbye agent')
except Exception as e: # pylint: disable=broad-except
@@ -77,7 +83,7 @@ def retry(func, exception_type, quit_event):
pass
def server_thread(sock, handler, quit_event):
def server_thread(sock, handle_conn, quit_event):
"""Run a server on the specified socket."""
log.debug('server thread started')
@@ -93,8 +99,9 @@ def server_thread(sock, handler, quit_event):
except StopIteration:
log.debug('server stopped')
break
with contextlib.closing(conn):
handle_connection(conn, handler)
# Handle connections from SSH concurrently.
threading.Thread(target=handle_conn,
kwargs=dict(conn=conn)).start()
log.debug('server thread stopped')
@@ -115,14 +122,23 @@ def serve(handler, sock_path=None, timeout=UNIX_SOCKET_TIMEOUT):
If no connection is made during the specified timeout,
retry until the context is over.
"""
ssh_version = subprocess.check_output(['ssh', '-V'],
stderr=subprocess.STDOUT)
log.debug('local SSH version: %r', ssh_version)
if sock_path is None:
sock_path = tempfile.mktemp(prefix='ssh-agent-')
sock_path = tempfile.mktemp(prefix='trezor-ssh-agent-')
environ = {'SSH_AUTH_SOCK': sock_path, 'SSH_AGENT_PID': str(os.getpid())}
device_mutex = threading.Lock()
with unix_domain_socket_server(sock_path) as sock:
sock.settimeout(timeout)
quit_event = threading.Event()
kwargs = dict(sock=sock, handler=handler, quit_event=quit_event)
handle_conn = functools.partial(handle_connection,
handler=handler,
mutex=device_mutex)
kwargs = dict(sock=sock,
handle_conn=handle_conn,
quit_event=quit_event)
with spawn(server_thread, kwargs):
try:
yield environ

View File

@@ -3,7 +3,7 @@ import io
import mock
import pytest
from .. import client, factory, formats, util
from .. import client, device, formats, util
ADDR = [2147483661, 2810943954, 3938368396, 3454558782, 3848009040]
CURVE = 'nist256p1'
@@ -12,49 +12,23 @@ PUBKEY = (b'\x03\xd8(\xb5\xa6`\xbet0\x95\xac:[;]\xdc,\xbd\xdc?\xd7\xc0\xec'
b'\xdd\xbc+\xfar~\x9dAis')
PUBKEY_TEXT = ('ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzd'
'HAyNTYAAABBBNgotaZgvnQwlaw6Wztd3Cy93D/XwOzdvCv6cn6dQWlzNMEQeW'
'VUfhvrGljR2Z/CMRONY6ejB+9PnpUOPuzYqi8= ssh://localhost:22\n')
'VUfhvrGljR2Z/CMRONY6ejB+9PnpUOPuzYqi8= <localhost:22|nist256p1>\n')
class FakeConnection(object):
class MockDevice(device.interface.Device): # pylint: disable=abstract-method
def __init__(self):
self.closed = False
def connect(self): # pylint: disable=no-self-use
return mock.Mock()
def close(self):
self.closed = True
def pubkey(self, identity, ecdh=False): # pylint: disable=unused-argument
assert self.conn
return PUBKEY
def clear_session(self):
self.closed = True
def get_public_node(self, n, ecdsa_curve_name=b'secp256k1'):
assert not self.closed
assert n == ADDR
assert ecdsa_curve_name in {'secp256k1', 'nist256p1'}
result = mock.Mock(spec=[])
result.node = mock.Mock(spec=[])
result.node.public_key = PUBKEY
return result
def ping(self, msg):
assert not self.closed
return msg
def identity_type(**kwargs):
result = mock.Mock(spec=[])
result.index = 0
result.proto = result.user = result.host = result.port = None
result.path = None
for k, v in kwargs.items():
setattr(result, k, v)
return result
def load_client():
return factory.ClientWrapper(connection=FakeConnection(),
identity_type=identity_type,
device_name='DEVICE_NAME',
call_exception=Exception)
def sign(self, identity, blob):
"""Sign given blob and return the signature (as bytes)."""
assert self.conn
assert blob == BLOB
return SIG
BLOB = (b'\x00\x00\x00 \xce\xe0\xc9\xd5\xceu/\xe8\xc5\xf2\xbfR+x\xa1\xcf\xb0'
@@ -66,71 +40,33 @@ BLOB = (b'\x00\x00\x00 \xce\xe0\xc9\xd5\xceu/\xe8\xc5\xf2\xbfR+x\xa1\xcf\xb0'
b'\xdd\xbc+\xfar~\x9dAis4\xc1\x10yeT~\x1b\xeb\x1aX\xd1\xd9\x9f\xc21'
b'\x13\x8dc\xa7\xa3\x07\xefO\x9e\x95\x0e>\xec\xd8\xaa/')
SIG = (b'\x00R\x19T\xf2\x84$\xef#\x0e\xee\x04X\xc6\xc3\x99T`\xd1\xd8\xf7!'
SIG = (b'R\x19T\xf2\x84$\xef#\x0e\xee\x04X\xc6\xc3\x99T`\xd1\xd8\xf7!'
b'\x862@cx\xb8\xb9i@1\x1b3#\x938\x86]\x97*Y\xb2\x02Xa\xdf@\xecK'
b'\xdc\xf0H\xab\xa8\xac\xa7? \x8f=C\x88N\xe2')
def test_ssh_agent():
label = 'localhost:22'
c = client.Client(loader=load_client)
ident = c.get_identity(label=label)
assert ident.host == 'localhost'
assert ident.proto == 'ssh'
assert ident.port == '22'
assert ident.user is None
assert ident.path is None
assert ident.index == 0
identity = device.interface.Identity(identity_str='localhost:22',
curve_name=CURVE)
c = client.Client(device=MockDevice())
assert c.get_public_key(identity) == PUBKEY_TEXT
signature = c.sign_ssh_challenge(blob=BLOB, identity=identity)
with c:
assert c.get_public_key(label) == PUBKEY_TEXT
key = formats.import_public_key(PUBKEY_TEXT)
serialized_sig = key['verifier'](sig=signature, msg=BLOB)
def ssh_sign_identity(identity, challenge_hidden,
challenge_visual, ecdsa_curve_name):
assert (client.identity_to_string(identity) ==
client.identity_to_string(ident))
assert challenge_hidden == BLOB
assert challenge_visual == ''
assert ecdsa_curve_name == 'nist256p1'
stream = io.BytesIO(serialized_sig)
r = util.read_frame(stream)
s = util.read_frame(stream)
assert not stream.read()
assert r[:1] == b'\x00'
assert s[:1] == b'\x00'
assert r[1:] + s[1:] == SIG
result = mock.Mock(spec=[])
result.public_key = PUBKEY
result.signature = SIG
return result
# pylint: disable=unused-argument
def cancel_sign(identity, blob):
raise IOError(42, 'ERROR')
c.client.sign_identity = ssh_sign_identity
signature = c.sign_ssh_challenge(label=label, blob=BLOB)
key = formats.import_public_key(PUBKEY_TEXT)
serialized_sig = key['verifier'](sig=signature, msg=BLOB)
stream = io.BytesIO(serialized_sig)
r = util.read_frame(stream)
s = util.read_frame(stream)
assert not stream.read()
assert r[:1] == b'\x00'
assert s[:1] == b'\x00'
assert r[1:] + s[1:] == SIG[1:]
c.client.call_exception = ValueError
# pylint: disable=unused-argument
def cancel_sign_identity(identity, challenge_hidden,
challenge_visual, ecdsa_curve_name):
raise c.client.call_exception(42, 'ERROR')
c.client.sign_identity = cancel_sign_identity
with pytest.raises(IOError):
c.sign_ssh_challenge(label=label, blob=BLOB)
def test_utils():
identity = mock.Mock(spec=[])
identity.proto = 'https'
identity.user = 'user'
identity.host = 'host'
identity.port = '443'
identity.path = '/path'
url = 'https://user@host:443/path'
assert client.identity_to_string(identity) == url
c.device.sign = cancel_sign
with pytest.raises(IOError):
c.sign_ssh_challenge(blob=BLOB, identity=identity)

View File

@@ -1,97 +0,0 @@
import mock
import pytest
from .. import factory
def test_load():
def single():
return [0]
def nothing():
return []
def double():
return [1, 2]
assert factory.load(loaders=[single]) == 0
assert factory.load(loaders=[single, nothing]) == 0
assert factory.load(loaders=[nothing, single]) == 0
with pytest.raises(IOError):
factory.load(loaders=[])
with pytest.raises(IOError):
factory.load(loaders=[single, single])
with pytest.raises(IOError):
factory.load(loaders=[double])
def factory_load_client(**kwargs):
# pylint: disable=protected-access
return list(factory._load_client(**kwargs))
def test_load_nothing():
hid_transport = mock.Mock(spec_set=['enumerate'])
hid_transport.enumerate.return_value = []
result = factory_load_client(
name=None,
client_type=None,
hid_transport=hid_transport,
passphrase_ack=None,
identity_type=None,
required_version=None,
call_exception=None)
assert result == []
def create_client_type(version):
conn = mock.Mock(spec=[])
conn.features = mock.Mock(spec=[])
major, minor, patch = version.split('.')
conn.features.device_id = 'DEVICE_ID'
conn.features.label = 'LABEL'
conn.features.vendor = 'VENDOR'
conn.features.major_version = major
conn.features.minor_version = minor
conn.features.patch_version = patch
conn.features.revision = b'\x12\x34\x56\x78'
return mock.Mock(spec_set=[], return_value=conn)
def test_load_single():
hid_transport = mock.Mock(spec_set=['enumerate'])
hid_transport.enumerate.return_value = [0]
for version in ('1.3.4', '1.3.5', '1.4.0', '2.0.0'):
passphrase_ack = mock.Mock(spec_set=[])
client_type = create_client_type(version)
client_wrapper, = factory_load_client(
name='DEVICE_NAME',
client_type=client_type,
hid_transport=hid_transport,
passphrase_ack=passphrase_ack,
identity_type=None,
required_version='>=1.3.4',
call_exception=None)
assert client_wrapper.connection is client_type.return_value
assert client_wrapper.device_name == 'DEVICE_NAME'
client_wrapper.connection.callback_PassphraseRequest('MESSAGE')
assert passphrase_ack.mock_calls == [mock.call(passphrase='')]
def test_load_old():
hid_transport = mock.Mock(spec_set=['enumerate'])
hid_transport.enumerate.return_value = [0]
for version in ('1.3.3', '1.2.5', '1.1.0', '0.9.9'):
with pytest.raises(ValueError):
factory_load_client(
name='DEVICE_NAME',
client_type=create_client_type(version),
hid_transport=hid_transport,
passphrase_ack=None,
identity_type=None,
required_version='>=1.3.4',
call_exception=None)

View File

@@ -93,3 +93,11 @@ def test_curve_mismatch():
def test_serialize_error():
with pytest.raises(TypeError):
formats.serialize_verifying_key(None)
def test_get_ecdh_curve_name():
for c in [formats.CURVE_NIST256, formats.ECDH_CURVE25519]:
assert c == formats.get_ecdh_curve_name(c)
assert (formats.ECDH_CURVE25519 ==
formats.get_ecdh_curve_name(formats.CURVE_ED25519))

View File

@@ -1,6 +1,6 @@
import pytest
from .. import formats, protocol
from .. import device, formats, protocol
# pylint: disable=line-too-long
@@ -17,6 +17,7 @@ NIST256_SIGN_REPLY = b'\x00\x00\x00j\x0e\x00\x00\x00e\x00\x00\x00\x13ecdsa-sha2-
def test_list():
key = formats.import_public_key(NIST256_KEY)
key['identity'] = device.interface.Identity('ssh://localhost', 'nist256p1')
h = protocol.Handler(keys=[key], signer=None)
reply = h.handle(LIST_MSG)
assert reply == LIST_NIST256_REPLY
@@ -28,14 +29,15 @@ def test_unsupported():
assert reply == b'\x00\x00\x00\x01\x05'
def ecdsa_signer(label, blob):
assert label == 'ssh://localhost'
def ecdsa_signer(identity, blob):
assert str(identity) == '<ssh://localhost|nist256p1>'
assert blob == NIST256_BLOB
return NIST256_SIG
def test_ecdsa_sign():
key = formats.import_public_key(NIST256_KEY)
key['identity'] = device.interface.Identity('ssh://localhost', 'nist256p1')
h = protocol.Handler(keys=[key], signer=ecdsa_signer)
reply = h.handle(NIST256_SIGN_MSG)
assert reply == NIST256_SIGN_REPLY
@@ -43,31 +45,30 @@ def test_ecdsa_sign():
def test_sign_missing():
h = protocol.Handler(keys=[], signer=ecdsa_signer)
with pytest.raises(KeyError):
h.handle(NIST256_SIGN_MSG)
def test_sign_wrong():
def wrong_signature(label, blob):
assert label == 'ssh://localhost'
def wrong_signature(identity, blob):
assert str(identity) == '<ssh://localhost|nist256p1>'
assert blob == NIST256_BLOB
return b'\x00' * 64
key = formats.import_public_key(NIST256_KEY)
key['identity'] = device.interface.Identity('ssh://localhost', 'nist256p1')
h = protocol.Handler(keys=[key], signer=wrong_signature)
with pytest.raises(ValueError):
h.handle(NIST256_SIGN_MSG)
def test_sign_cancel():
def cancel_signature(label, blob): # pylint: disable=unused-argument
def cancel_signature(identity, blob): # pylint: disable=unused-argument
raise IOError()
key = formats.import_public_key(NIST256_KEY)
key['identity'] = device.interface.Identity('ssh://localhost', 'nist256p1')
h = protocol.Handler(keys=[key], signer=cancel_signature)
assert h.handle(NIST256_SIGN_MSG) == protocol.failure()
@@ -79,14 +80,15 @@ ED25519_BLOB = b'''\x00\x00\x00 i3\xae}yk\\\xa1L\xb9\xe1\xbf\xbc\x8e\x87\r\x0e\x
ED25519_SIG = b'''\x8eb)\xa6\xe9P\x83VE\xfbq\xc6\xbf\x1dV3\xe3<O\x11\xc0\xfa\xe4\xed\xb8\x81.\x81\xc8\xa6\xba\x10RA'a\xbc\xa9\xd3\xdb\x98\x07\xf0\x1a\x9c4\x84<\xaf\x99\xb7\xe5G\xeb\xf7$\xc1\r\x86f\x16\x8e\x08\x05''' # nopep8
def ed25519_signer(label, blob):
assert label == 'ssh://localhost'
def ed25519_signer(identity, blob):
assert str(identity) == '<ssh://localhost|ed25519>'
assert blob == ED25519_BLOB
return ED25519_SIG
def test_ed25519_sign():
key = formats.import_public_key(ED25519_KEY)
key['identity'] = device.interface.Identity('ssh://localhost', 'ed25519')
h = protocol.Handler(keys=[key], signer=ed25519_signer)
reply = h.handle(ED25519_SIGN_MSG)
assert reply == ED25519_SIGN_REPLY

View File

@@ -1,3 +1,4 @@
import functools
import io
import os
import socket
@@ -37,30 +38,32 @@ class FakeSocket(object):
def test_handle():
mutex = threading.Lock()
handler = protocol.Handler(keys=[], signer=None)
conn = FakeSocket()
server.handle_connection(conn, handler)
server.handle_connection(conn, handler, mutex)
msg = bytearray([protocol.msg_code('SSH_AGENTC_REQUEST_RSA_IDENTITIES')])
conn = FakeSocket(util.frame(msg))
server.handle_connection(conn, handler)
server.handle_connection(conn, handler, mutex)
assert conn.tx.getvalue() == b'\x00\x00\x00\x05\x02\x00\x00\x00\x00'
msg = bytearray([protocol.msg_code('SSH2_AGENTC_REQUEST_IDENTITIES')])
conn = FakeSocket(util.frame(msg))
server.handle_connection(conn, handler)
server.handle_connection(conn, handler, mutex)
assert conn.tx.getvalue() == b'\x00\x00\x00\x05\x0C\x00\x00\x00\x00'
msg = bytearray([protocol.msg_code('SSH2_AGENTC_ADD_IDENTITY')])
conn = FakeSocket(util.frame(msg))
server.handle_connection(conn, handler)
server.handle_connection(conn, handler, mutex)
conn.tx.seek(0)
reply = util.read_frame(conn.tx)
assert reply == util.pack('B', protocol.msg_code('SSH_AGENT_FAILURE'))
conn_mock = mock.Mock(spec=FakeSocket)
conn_mock.recv.side_effect = [Exception, EOFError]
server.handle_connection(conn=conn_mock, handler=None)
server.handle_connection(conn=conn_mock, handler=None, mutex=mutex)
def test_server_thread():
@@ -78,8 +81,10 @@ def test_server_thread():
def getsockname(self): # pylint: disable=no-self-use
return 'fake_server'
handler = protocol.Handler(keys=[], signer=None),
handle_conn = functools.partial(server.handle_connection, handler=handler)
server.server_thread(sock=FakeServer(),
handler=protocol.Handler(keys=[], signer=None),
handle_conn=handle_conn,
quit_event=quit_event)

View File

@@ -97,3 +97,7 @@ def test_reader():
with pytest.raises(EOFError):
r.read(1)
def test_setup_logging():
util.setup_logging(verbosity=10)

View File

@@ -2,8 +2,11 @@
import binascii
import contextlib
import io
import logging
import struct
log = logging.getLogger(__name__)
def send(conn, data):
"""Send data blob to connection socket."""
@@ -173,3 +176,12 @@ class Reader(object):
yield
finally:
self._captured = None
def setup_logging(verbosity, **kwargs):
"""Configure logging for this tool."""
fmt = ('%(asctime)s %(levelname)-12s %(message)-100s '
'[%(filename)s:%(lineno)d]')
levels = [logging.WARNING, logging.INFO, logging.DEBUG]
level = levels[min(verbosity, len(levels) - 1)]
logging.basicConfig(format=fmt, level=level, **kwargs)