Add first mostly working version

This commit is contained in:
IamTheFij 2017-08-07 10:27:29 -07:00
parent b5faab8870
commit 0b599cfd37
13 changed files with 850 additions and 0 deletions

26
.pre-commit-config.yaml Normal file
View File

@ -0,0 +1,26 @@
- repo: git://github.com/pre-commit/pre-commit-hooks
sha: 1553c96e2a0d0154f3aca4c5cb0156a74a8c703d
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
- id: autopep8-wrapper
exclude: ./src/vendor/.+
args:
- -i
- --ignore=E265,E309,E501
- id: debug-statements
exclude: ./src/vendor/.+
language_version: python2.7
- id: flake8
exclude: ./src/vendor/.+
language_version: python2.7
- id: check-yaml
- id: check-json
- id: name-tests-test
exclude: tests/(common.py|util.py|(helpers)/(.+).py)
- repo: git://github.com/asottile/reorder_python_imports
sha: ab609b9b982729dfc287b4e75963c0c4de254a31
hooks:
- id: reorder-python-imports
exclude: ./src/vendor/.+
language_version: python2.7

52
Makefile Normal file
View File

@ -0,0 +1,52 @@
.PHONY: default
default: run
# Simple execution of the workflow to see all results
.PHONY: run
run: install-requirements
./src/main.py
# Runs workflow and prompts for Yubikey password
.PHONY: set-password
set-password: install-requirements
./src/main.py set-password
.PHONY: install-requirements
install-requirements: src/vendor src/libusb-1.0.dylib
# Installs libusb from /opt/local, where MacPorts installs it
src/libusb-1.0.dylib:
cp /opt/local/lib/libusb-1.0.dylib ./src/
# Installs 3rd party packages into vendor directory
src/vendor:
mkdir -p .pip-cache
pip install -r ./requirements.txt -t ./src/vendor --cache-dir .pip-cache
cp vendor.py src/vendor/__init__.py
# Creates virtualenv for testing using MacPorts Python
virtualenv:
virtualenv --python=/opt/local/bin/python2.7 virtualenv
./virtualenv_run/bin/pip install -r ./requirements.txt
# Runs workflow using virtualenv Python
.PHONY: virtualenv_run
virtualenv_run: virtualenv
./virtualenv/bin/python src/main.py
# Clears the virtualenv and other installed files
.PHONY: clean
clean:
rm -fr virtualenv src/vendor src/libusb-1.0.dylib
find . -name '*.pyc' -delete
find . -name '__pycache__' -delete
# Installs required MacPorts
.PHONY: install-ports
install-ports:
sudo port install swig swig-python ykpers libu2f-host libusb
# Install precommit hooks
.PHONY: intall-hooks
install-hooks:
tox -e pre-commit -- install -f --install-hooks

2
requirements.txt Normal file
View File

@ -0,0 +1,2 @@
alfred-workflow==1.27
yubikey-manager==0.4.0

4
set_version.sh Executable file
View File

@ -0,0 +1,4 @@
#! /bin/bash
echo "$1" > ./src/version
plutil -replace version -string "$1" ./src/info.plist

32
setup.py Normal file
View File

@ -0,0 +1,32 @@
#! /usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import absolute_import
from __future__ import unicode_literals
import os
from setuptools import find_packages
from setuptools import setup
def read_version():
version_path = 'version'
try:
return open(
os.path.join(os.path.dirname(__file__), version_path)
).read()
except:
pass
setup(
name='Yubico Auth Workflow',
version=read_version(),
description='Yubico Auth workflow for Alfred',
author='Ian Fijolek',
author_email='ian@iamthefij.com.com',
url='',
packages=find_packages(exclude=['tests*']),
install_requires=[],
license='MIT',
)

Binary file not shown.

After

Width:  |  Height:  |  Size: 239 KiB

237
src/controller.py Normal file
View File

@ -0,0 +1,237 @@
import hashlib
from binascii import a2b_hex
from binascii import b2a_hex
from ykman.descriptor import get_descriptors
from ykman.driver_ccid import APDUError
from ykman.driver_otp import YkpersError
from ykman.oath import Credential
from ykman.oath import OathController
from ykman.oath import SW
from ykman.util import CAPABILITY
from ykman.util import derive_key
from ykman.util import parse_b32_key
from ykman.util import TRANSPORT
import vendor # noqa
NON_FEATURE_CAPABILITIES = [CAPABILITY.CCID, CAPABILITY.NFC]
class DeviceNotFoundError(Exception):
pass
class CouldNotOpenDeviceError(Exception):
pass
class Controller(object):
_descriptor = None
_dev_info = None
def get_features(self):
return [
c.name for c in CAPABILITY if c not in NON_FEATURE_CAPABILITIES]
def count_devices(self):
return len(list(get_descriptors()))
def refresh(self):
descriptors = list(get_descriptors())
if len(descriptors) != 1:
self._descriptor = None
raise DeviceNotFoundError()
return
desc = descriptors[0]
if desc.fingerprint != (
self._descriptor.fingerprint if self._descriptor else None):
dev = desc.open_device()
if not dev:
raise CouldNotOpenDeviceError()
return
self._descriptor = desc
self._dev_info = {
'name': dev.device_name,
'version': '.'.join(str(x) for x in dev.version),
'serial': dev.serial or '',
'enabled': [c.name for c in CAPABILITY if c & dev.enabled],
'connections': [
t.name for t in TRANSPORT if t & dev.capabilities],
}
return self._dev_info
def refresh_credentials(self, timestamp, password_key=None):
return [
c.to_dict() for c in self._calculate_all(timestamp, password_key)]
def calculate(self, credential, timestamp, password_key):
return self._calculate(
Credential.from_dict(
credential), timestamp, password_key).to_dict()
def calculate_slot_mode(self, slot, digits, timestamp):
dev = self._descriptor.open_device(TRANSPORT.OTP)
code = dev.driver.calculate(
slot, challenge=timestamp, totp=True, digits=int(digits),
wait_for_touch=True)
return Credential(
self._slot_name(slot), code=code, oath_type='totp', touch=True,
algo='SHA1', expiration=self._expiration(timestamp)).to_dict()
def refresh_slot_credentials(self, slots, digits, timestamp):
result = []
if slots[0]:
cred = self._read_slot_cred(1, digits[0], timestamp)
if cred:
result.append(cred)
if slots[1]:
cred = self._read_slot_cred(2, digits[1], timestamp)
if cred:
result.append(cred)
return [c.to_dict() for c in result]
def _read_slot_cred(self, slot, digits, timestamp):
try:
dev = self._descriptor.open_device(TRANSPORT.OTP)
code = dev.driver.calculate(
slot, challenge=timestamp, totp=True, digits=int(digits),
wait_for_touch=False)
return Credential(
self._slot_name(slot), code=code, oath_type='totp',
touch=False, algo='SHA1',
expiration=self._expiration(timestamp))
except YkpersError as e:
if e.errno == 11:
return Credential(
self._slot_name(slot), oath_type='totp', touch=True,
algo='SHA1')
except:
pass
return None
def _slot_name(self, slot):
return "YubiKey Slot {}".format(slot)
def _expiration(self, timestamp):
return ((timestamp + 30) // 30) * 30
def needs_validation(self):
try:
dev = self._descriptor.open_device(TRANSPORT.CCID)
controller = OathController(dev.driver)
return controller.locked
except:
return False
def get_oath_id(self):
dev = self._descriptor.open_device(TRANSPORT.CCID)
controller = OathController(dev.driver)
return b2a_hex(controller.id).decode('utf-8')
def derive_key(self, password):
dev = self._descriptor.open_device(TRANSPORT.CCID)
controller = OathController(dev.driver)
key = derive_key(controller.id, password)
return b2a_hex(key).decode('utf-8')
def validate(self, key):
dev = self._descriptor.open_device(TRANSPORT.CCID)
controller = OathController(dev.driver)
if key is not None:
try:
controller.validate(a2b_hex(key))
return True
except:
return False
def set_password(self, new_password, password_key):
dev = self._descriptor.open_device(TRANSPORT.CCID)
controller = OathController(dev.driver)
if controller.locked and password_key is not None:
controller.validate(a2b_hex(password_key))
if new_password is not None:
key = derive_key(controller.id, new_password)
controller.set_password(key)
else:
controller.clear_password()
def add_credential(
self, name, key, oath_type, digits, algo, touch, password_key):
dev = self._descriptor.open_device(TRANSPORT.CCID)
controller = OathController(dev.driver)
if controller.locked and password_key is not None:
controller.validate(a2b_hex(password_key))
try:
key = parse_b32_key(key)
except Exception as e:
return str(e)
try:
controller.put(
key, name, oath_type, digits, algo=algo, require_touch=touch)
except APDUError as e:
# NEO doesn't return a no space error if full,
# but a command aborted error. Assume it's because of
# no space in this context.
if e.sw == SW.NO_SPACE or e.sw == SW.COMMAND_ABORTED:
return 'No space'
else:
raise
def add_slot_credential(self, slot, key, touch):
dev = self._descriptor.open_device(TRANSPORT.OTP)
key = parse_b32_key(key)
if len(key) > 64: # Keys longer than 64 bytes are hashed.
key = hashlib.sha1(key).digest()
if len(key) > 20:
raise ValueError(
'YubiKey Slots cannot handle TOTP keys over 20 bytes.')
key += b'\x00' * (20 - len(key)) # Keys must be padded to 20 bytes.
dev.driver.program_chalresp(int(slot), key, touch)
def delete_slot_credential(self, slot):
dev = self._descriptor.open_device(TRANSPORT.OTP)
dev.driver.zap_slot(slot)
def delete_credential(self, credential, password_key):
dev = self._descriptor.open_device(TRANSPORT.CCID)
controller = OathController(dev.driver)
if controller.locked and password_key is not None:
controller.validate(a2b_hex(password_key))
controller.delete(Credential.from_dict(credential))
def _calculate(self, credential, timestamp, password_key):
dev = self._descriptor.open_device(TRANSPORT.CCID)
controller = OathController(dev.driver)
if controller.locked and password_key is not None:
controller.validate(a2b_hex(password_key))
cred = controller.calculate(credential, timestamp)
return cred
def _calculate_all(self, timestamp, password_key):
dev = self._descriptor.open_device(TRANSPORT.CCID)
controller = OathController(dev.driver)
if controller.locked and password_key is not None:
controller.validate(a2b_hex(password_key))
creds = controller.calculate_all(timestamp)
creds = [c for c in creds if not c.hidden]
return creds
def reset(self):
dev = self._descriptor.open_device(TRANSPORT.CCID)
controller = OathController(dev.driver)
controller.reset()
def slot_status(self):
dev = self._descriptor.open_device(TRANSPORT.OTP)
return list(dev.driver.slot_status)
def list_credentials(self, password_key):
dev = self._descriptor.open_device(TRANSPORT.CCID)
controller = OathController(dev.driver)
if controller.locked and password_key is not None:
controller.validate(a2b_hex(password_key))
creds = controller.list()
return [c.to_dict() for c in creds]

BIN
src/icon.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 239 KiB

306
src/info.plist Normal file
View File

@ -0,0 +1,306 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<key>bundleid</key>
<string>com.vividboarder.alfred-yubico-auth</string>
<key>connections</key>
<dict>
<key>8486DCAA-AFB7-407D-A0E9-E57E09997B24</key>
<array>
<dict>
<key>destinationuid</key>
<string>9F48DDE6-BBE7-42ED-ABE5-C9255C92F1CD</string>
<key>modifiers</key>
<integer>0</integer>
<key>modifiersubtext</key>
<string></string>
<key>vitoclose</key>
<false/>
</dict>
<dict>
<key>destinationuid</key>
<string>FB0DDF80-FF90-439A-BF3F-6EC58C2AA870</string>
<key>modifiers</key>
<integer>1048576</integer>
<key>modifiersubtext</key>
<string>Paste token</string>
<key>vitoclose</key>
<false/>
</dict>
</array>
<key>9F48DDE6-BBE7-42ED-ABE5-C9255C92F1CD</key>
<array>
<dict>
<key>destinationuid</key>
<string>DA3E9CE8-7C4F-4B09-BFA5-F8CA83297968</string>
<key>modifiers</key>
<integer>0</integer>
<key>modifiersubtext</key>
<string></string>
<key>vitoclose</key>
<false/>
</dict>
</array>
<key>BFB3A122-52BB-4FF1-B5B3-CECD42A730DB</key>
<array>
<dict>
<key>destinationuid</key>
<string>0718204D-3398-4AEF-A621-DDDE1FC6ED75</string>
<key>modifiers</key>
<integer>0</integer>
<key>modifiersubtext</key>
<string></string>
<key>vitoclose</key>
<false/>
</dict>
</array>
<key>DA3E9CE8-7C4F-4B09-BFA5-F8CA83297968</key>
<array>
<dict>
<key>destinationuid</key>
<string>E86DC7C1-35C1-4BA9-8B33-A95DE7082F7E</string>
<key>modifiers</key>
<integer>0</integer>
<key>modifiersubtext</key>
<string></string>
<key>vitoclose</key>
<false/>
</dict>
</array>
<key>E86DC7C1-35C1-4BA9-8B33-A95DE7082F7E</key>
<array>
<dict>
<key>destinationuid</key>
<string>BFB3A122-52BB-4FF1-B5B3-CECD42A730DB</string>
<key>modifiers</key>
<integer>0</integer>
<key>modifiersubtext</key>
<string></string>
<key>vitoclose</key>
<false/>
</dict>
</array>
</dict>
<key>createdby</key>
<string>ViViDboarder</string>
<key>description</key>
<string>2FA for Yubikeys</string>
<key>disabled</key>
<false/>
<key>name</key>
<string>Yubico Auth</string>
<key>objects</key>
<array>
<dict>
<key>config</key>
<dict>
<key>lastpathcomponent</key>
<false/>
<key>onlyshowifquerypopulated</key>
<false/>
<key>removeextension</key>
<false/>
<key>text</key>
<string>Clipboard cleared!</string>
<key>title</key>
<string>Yubico Auth</string>
</dict>
<key>type</key>
<string>alfred.workflow.output.notification</string>
<key>uid</key>
<string>0718204D-3398-4AEF-A621-DDDE1FC6ED75</string>
<key>version</key>
<integer>1</integer>
</dict>
<dict>
<key>config</key>
<dict>
<key>autopaste</key>
<false/>
<key>clipboardtext</key>
<string>{query}</string>
<key>transient</key>
<true/>
</dict>
<key>type</key>
<string>alfred.workflow.output.clipboard</string>
<key>uid</key>
<string>9F48DDE6-BBE7-42ED-ABE5-C9255C92F1CD</string>
<key>version</key>
<integer>2</integer>
</dict>
<dict>
<key>config</key>
<dict>
<key>autopaste</key>
<false/>
<key>clipboardtext</key>
<string></string>
<key>transient</key>
<false/>
</dict>
<key>type</key>
<string>alfred.workflow.output.clipboard</string>
<key>uid</key>
<string>BFB3A122-52BB-4FF1-B5B3-CECD42A730DB</string>
<key>version</key>
<integer>2</integer>
</dict>
<dict>
<key>config</key>
<dict>
<key>lastpathcomponent</key>
<false/>
<key>onlyshowifquerypopulated</key>
<false/>
<key>removeextension</key>
<false/>
<key>text</key>
<string>Coppied!</string>
<key>title</key>
<string>Yubico Auth</string>
</dict>
<key>type</key>
<string>alfred.workflow.output.notification</string>
<key>uid</key>
<string>DA3E9CE8-7C4F-4B09-BFA5-F8CA83297968</string>
<key>version</key>
<integer>1</integer>
</dict>
<dict>
<key>config</key>
<dict>
<key>alfredfiltersresults</key>
<true/>
<key>argumenttrimmode</key>
<integer>0</integer>
<key>argumenttype</key>
<integer>1</integer>
<key>escaping</key>
<integer>0</integer>
<key>keyword</key>
<string>yubikey</string>
<key>queuedelaycustom</key>
<integer>3</integer>
<key>queuedelayimmediatelyinitially</key>
<true/>
<key>queuedelaymode</key>
<integer>0</integer>
<key>queuemode</key>
<integer>1</integer>
<key>runningsubtext</key>
<string></string>
<key>script</key>
<string>python main.py</string>
<key>scriptargtype</key>
<integer>1</integer>
<key>scriptfile</key>
<string>main.py</string>
<key>subtext</key>
<string>Get 2FA tokens from Yubikey</string>
<key>title</key>
<string>Yubikey 2FA</string>
<key>type</key>
<integer>8</integer>
<key>withspace</key>
<true/>
</dict>
<key>type</key>
<string>alfred.workflow.input.scriptfilter</string>
<key>uid</key>
<string>8486DCAA-AFB7-407D-A0E9-E57E09997B24</string>
<key>version</key>
<integer>2</integer>
</dict>
<dict>
<key>config</key>
<dict>
<key>seconds</key>
<string>10</string>
</dict>
<key>type</key>
<string>alfred.workflow.utility.delay</string>
<key>uid</key>
<string>E86DC7C1-35C1-4BA9-8B33-A95DE7082F7E</string>
<key>version</key>
<integer>1</integer>
</dict>
<dict>
<key>config</key>
<dict>
<key>autopaste</key>
<true/>
<key>clipboardtext</key>
<string>{query}</string>
<key>transient</key>
<true/>
</dict>
<key>type</key>
<string>alfred.workflow.output.clipboard</string>
<key>uid</key>
<string>FB0DDF80-FF90-439A-BF3F-6EC58C2AA870</string>
<key>version</key>
<integer>2</integer>
</dict>
</array>
<key>readme</key>
<string></string>
<key>uidata</key>
<dict>
<key>0718204D-3398-4AEF-A621-DDDE1FC6ED75</key>
<dict>
<key>xpos</key>
<integer>960</integer>
<key>ypos</key>
<integer>140</integer>
</dict>
<key>8486DCAA-AFB7-407D-A0E9-E57E09997B24</key>
<dict>
<key>xpos</key>
<integer>120</integer>
<key>ypos</key>
<integer>140</integer>
</dict>
<key>9F48DDE6-BBE7-42ED-ABE5-C9255C92F1CD</key>
<dict>
<key>xpos</key>
<integer>360</integer>
<key>ypos</key>
<integer>140</integer>
</dict>
<key>BFB3A122-52BB-4FF1-B5B3-CECD42A730DB</key>
<dict>
<key>xpos</key>
<integer>780</integer>
<key>ypos</key>
<integer>140</integer>
</dict>
<key>DA3E9CE8-7C4F-4B09-BFA5-F8CA83297968</key>
<dict>
<key>xpos</key>
<integer>540</integer>
<key>ypos</key>
<integer>140</integer>
</dict>
<key>E86DC7C1-35C1-4BA9-8B33-A95DE7082F7E</key>
<dict>
<key>xpos</key>
<integer>700</integer>
<key>ypos</key>
<integer>170</integer>
</dict>
<key>FB0DDF80-FF90-439A-BF3F-6EC58C2AA870</key>
<dict>
<key>xpos</key>
<integer>360</integer>
<key>ypos</key>
<integer>290</integer>
</dict>
</dict>
<key>version</key>
<string>0.0.1</string>
<key>webaddress</key>
<string></string>
</dict>
</plist>

154
src/main.py Executable file
View File

@ -0,0 +1,154 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys
from getpass import getpass
from time import time
from controller import Controller
from controller import DeviceNotFoundError
from workflow import ICON_ACCOUNT
from workflow import ICON_ERROR
from workflow import Workflow3
YUBIKEY_CREDS_KEYCHAIN = 'yubico-auth-creds'
def cred_to_item_kwargs(cred):
if cred.get('hidden'):
return None
return {
'icon': ICON_ACCOUNT,
'title': cred['name'],
'subtitle': 'Copy to clipboard',
'copytext': cred['code'],
'arg': cred['code'],
'valid': True,
}
class YubicoAuth(Workflow3):
_controller = None
def get_controller(self):
if not self._controller:
self._controller = Controller()
self._controller.refresh()
return self._controller
def ask_yubikey_password(self):
"""Prompts the user for their Yubikey password and stores it"""
self.logger.debug('Set password')
password_key = self.get_controller().derive_key(getpass())
self.save_password(YUBIKEY_CREDS_KEYCHAIN, password_key)
def get_yubikey_password(self):
"""Returns stored Yubikey password from keychain"""
return self.get_password(YUBIKEY_CREDS_KEYCHAIN)
def _get_positional_arg(self, position):
"""Safely return a positional argument"""
if len(self.args) > position:
return self.args[position]
return None
def get_command(self):
"""Get command out of the args as first parameter"""
return self._get_positional_arg(0)
def get_query(self):
"""Get query out of the args after first parameter"""
if len(self.args) < 2:
return None
return ' '.join(self.args[1:])
def _validate(self, command):
"""Validates that we can handle the current command"""
# if self.get_api_key() is None:
# self.add_item(
# title='Missing API key',
# subtitle='Set variable in settings',
# icon=ICON_ACCOUNT,
# valid=False,
# )
# return False
# if command == COMMAND_LOVE and self.get_recipient() is None:
# self.add_item(
# title='Recipient is required',
# icon=ICON_ERROR,
# valid=False,
# )
# return False
return True
def _add_cred_to_results(self, cred):
self.logger.debug('Read {}'.format(cred.get('name')))
item_args = cred_to_item_kwargs(cred)
if item_args:
self.add_item(**item_args)
def list_credentials(self):
password_key = self.get_yubikey_password()
for cred in self.get_controller().list_credentials(password_key):
self._add_cred_to_results(cred)
def refresh_credentials(self):
key = self.get_yubikey_password()
for cred in self.get_controller().refresh_credentials(time(), key):
self._add_cred_to_results(cred)
def main(self):
self.logger.debug('Starting...')
command = self.get_command()
if not self._validate(command):
self.send_feedback()
return
command_action = None
if command == 'set-password':
command_action = self.ask_yubikey_password
if command == 'list':
command_action = self.list_credentials
else:
command_action = self.refresh_credentials
try:
command_action()
except DeviceNotFoundError:
self.add_item(
'Could not find device',
'Is your Yubikey plugged in?',
icon=ICON_ERROR,
)
self.send_feedback()
def no_wf():
controller = Controller()
print(controller.get_features())
print(controller.count_devices())
print(controller.refresh())
# TODO: Accept password in keychain, create special command for adding key
password = getpass('YubiKey password? ')
password_key = controller.derive_key(password)
timestamp = time()
print(controller.refresh_credentials(timestamp, password_key))
creds = controller.list_credentials(password_key)
print(creds)
def main(wf=None):
if wf is None:
no_wf()
else:
wf.main()
if __name__ == '__main__':
# main()
wf = YubicoAuth()
sys.exit(wf.run(main))

1
src/version Normal file
View File

@ -0,0 +1 @@
0.0.1

31
tox.ini Normal file
View File

@ -0,0 +1,31 @@
[tox]
envlist = py
indexserver =
default = https://pypi.python.org/simple/
[testenv]
deps = -rrequirements-dev.txt
[testenv:py]
deps = {[testenv]deps}
commands =
coverage run --source=./src/,tests/ -m pytest --strict {posargs}
coverage report -m
pre-commit run --all-files
[testenv:lint]
deps = {[testenv]deps}
flake8
commands = flake8 .
[testenv:pre-commit]
deps = pre-commit>=0.4.2
commands = pre-commit {posargs}
[flake8]
exclude = .svn,CVS,.bzr,.hg,.git,__pycache__,.ropeproject,.tox,docs,virtualenv_run
filename = *.py,*.wsgi
max-line-length = 80
[pytest]
norecursedirs = .* _darcs CVS docs virtualenv_run

5
vendor.py Normal file
View File

@ -0,0 +1,5 @@
import os
import sys
sys.path.append(os.path.basename(__file__))