Added a sample DBus service which can impersoname fprintd
This commit is contained in:
parent
ef6ce24a26
commit
b690db7498
3 changed files with 160 additions and 19 deletions
132
dbus-service.py
Normal file
132
dbus-service.py
Normal file
|
|
@ -0,0 +1,132 @@
|
|||
|
||||
from threading import Thread
|
||||
from gi.repository import GLib
|
||||
from pydbus import SystemBus
|
||||
from pydbus.generic import signal
|
||||
import pkg_resources
|
||||
from time import sleep
|
||||
from prototype import *
|
||||
import pwd
|
||||
|
||||
load97()
|
||||
|
||||
loop = GLib.MainLoop()
|
||||
|
||||
def uname2identity(uname):
|
||||
if uname == '':
|
||||
# For some reason Gnome enrollment UI does not send the user name
|
||||
# (it also ignores our num-enroll-stages attribute). I probably should upgrade Gnome
|
||||
print('No username specified. ')
|
||||
uname = 'unicorn'
|
||||
pw=pwd.getpwnam(uname)
|
||||
sidstr='S-1-5-21-111111111-1111111111-1111111111-%d' % pw.pw_uid
|
||||
return sid_from_string(sidstr)
|
||||
|
||||
|
||||
class Device():
|
||||
name = 'Validity sensor'
|
||||
|
||||
def __init__(self):
|
||||
setattr(self, 'num-enroll-stages', 7)
|
||||
setattr(self, 'scan-type', 'press')
|
||||
|
||||
def Claim(self, usr):
|
||||
print('In Claim %s' % usr)
|
||||
|
||||
def Release(self):
|
||||
print('In Release')
|
||||
|
||||
def ListEnrolledFingers(self, usr):
|
||||
print('In ListEnrolledFingers %s' % usr)
|
||||
|
||||
usr=db.lookup_user(uname2identity(usr))
|
||||
|
||||
if usr == None:
|
||||
print('User not found on this device')
|
||||
return []
|
||||
|
||||
rc = [subtype_to_string(f['subtype']) for f in usr.fingers]
|
||||
print(repr(rc))
|
||||
return rc
|
||||
|
||||
def do_scan(self):
|
||||
try:
|
||||
z=identify()
|
||||
self.VerifyStatus('verify-match', True)
|
||||
except Exception as e:
|
||||
print('Opps: %s' % repr(e))
|
||||
self.VerifyStatus('verify-no-match', True)
|
||||
|
||||
def VerifyStart(self, finger_name):
|
||||
print('In VerifyStart %s' % finger_name)
|
||||
Thread(target=lambda: self.do_scan()).start()
|
||||
|
||||
def VerifyStop(self):
|
||||
print('In VerifyStop')
|
||||
|
||||
def DeleteEnrolledFingers(self, user):
|
||||
print('In DeleteEnrolledFingers %s' % user)
|
||||
usr=db.lookup_user(uname2identity(user))
|
||||
|
||||
if usr == None:
|
||||
print('User not found on this device')
|
||||
return
|
||||
|
||||
db.del_record(usr.dbid)
|
||||
|
||||
def do_enroll(self, finger_name):
|
||||
# it is pointless to try and remember username passed in claim as Gnome does not seem to be passing anything useful anyway
|
||||
try:
|
||||
# hardcode the username and finger for now
|
||||
z=enroll(uname2identity('unicorn'), 0xf5)
|
||||
print('Enroll was successfull')
|
||||
self.EnrollStatus('enroll-completed', True)
|
||||
except Exception as e:
|
||||
print('Enroll failed %s' % repr(e))
|
||||
self.EnrollStatus('enroll-failed', True)
|
||||
|
||||
def EnrollStart(self, finger_name):
|
||||
print('In EnrollStart %s' % finger_name)
|
||||
Thread(target=lambda: self.do_enroll(finger_name)).start()
|
||||
|
||||
|
||||
def EnrollStop(self):
|
||||
print('In EnrollStop')
|
||||
|
||||
VerifyFingerSelected = signal()
|
||||
VerifyStatus = signal()
|
||||
EnrollStatus = signal()
|
||||
|
||||
class Manager():
|
||||
def GetDevices(self):
|
||||
print('In GetDevices')
|
||||
return ['/net/reactivated/Fprint/Device/0']
|
||||
|
||||
def GetDefaultDevice(self):
|
||||
print('In GetDefaultDevice')
|
||||
return '/net/reactivated/Fprint/Device/0'
|
||||
|
||||
|
||||
def readif(fn):
|
||||
with open('/usr/share/dbus-1/interfaces/' + fn, 'rb') as f:
|
||||
# for some reason Gio can't seem to handle XML entities declared inline
|
||||
return f.read().decode('utf-8') \
|
||||
.replace('&ERROR_CLAIM_DEVICE;', 'net.reactivated.Fprint.Error.ClaimDevice') \
|
||||
.replace('&ERROR_ALREADY_IN_USE;', 'net.reactivated.Fprint.Error.AlreadyInUse') \
|
||||
.replace('&ERROR_INTERNAL;', 'net.reactivated.Fprint.Error.Internal') \
|
||||
.replace('&ERROR_PERMISSION_DENIED;', 'net.reactivated.Fprint.Error.PermissionDenied') \
|
||||
.replace('&ERROR_NO_ENROLLED_PRINTS;', 'net.reactivated.Fprint.Error.NoEnrolledPrints') \
|
||||
.replace('&ERROR_NO_ACTION_IN_PROGRESS;', 'net.reactivated.Fprint.Error.NoActionInProgress') \
|
||||
.replace('&ERROR_INVALID_FINGERNAME;', 'net.reactivated.Fprint.Error.InvalidFingername') \
|
||||
.replace('&ERROR_NO_SUCH_DEVICE;', 'net.reactivated.Fprint.Error.NoSuchDevice')
|
||||
|
||||
Device.dbus=[readif('net.reactivated.Fprint.Device.xml')]
|
||||
Manager.dbus=[readif('net.reactivated.Fprint.Manager.xml')]
|
||||
|
||||
|
||||
bus = SystemBus()
|
||||
bus.publish('net.reactivated.Fprint',
|
||||
('/net/reactivated/Fprint/Manager', Manager()),
|
||||
('/net/reactivated/Fprint/Device/0', Device())
|
||||
)
|
||||
loop.run()
|
||||
|
|
@ -66,7 +66,7 @@ c1e1e1c221f201d21201e1c1f34242221201f20221f201e241e241d2020221e2420231d221e211e1
|
|||
08180818081808180808081808180808081807f808081808080818081808180808081808180818081808180818080808180818081808180
|
||||
81808180818081808180818081808180818081808080808080808080807f807f807f807f7f7e7e
|
||||
''')
|
||||
# The following blog has only 1 byte difference from the above
|
||||
# The following blob has only 1 byte difference from the above
|
||||
enroll_prg=unhex('''
|
||||
02980000002300000020000800002000800000010032007000000000802020050024200000502077362820010030200100082170000c210
|
||||
000482102004c210000582000005c20000060200000682005006c20012970200121742001887820018084202000942001809c200902a020
|
||||
|
|
|
|||
45
tls97.py
45
tls97.py
|
|
@ -11,6 +11,7 @@ from fastecdsa.point import Point
|
|||
from fastecdsa.keys import gen_private_key, get_public_key
|
||||
from fastecdsa.ecdsa import sign
|
||||
from fastecdsa.encoding.der import DEREncoder
|
||||
from util import assert_status
|
||||
import pickle
|
||||
|
||||
|
||||
|
|
@ -70,19 +71,32 @@ def prf(secret, seed, length):
|
|||
|
||||
return res[:length]
|
||||
|
||||
|
||||
# TODO assert the right state transitions
|
||||
class Tls():
|
||||
|
||||
def __init__(self, usb):
|
||||
self.usb = usb
|
||||
self.trace_enabled = False
|
||||
|
||||
def open(self):
|
||||
self.parse40(self.usb.cmd(unhexlify('40010100000000000000100000')))
|
||||
|
||||
self.secure_rx = False
|
||||
self.secure_tx = False
|
||||
|
||||
def read_flash(self, which, addr, size):
|
||||
cmd = pack('<BBBHLL', 0x40, which, 1, 0, addr, size)
|
||||
if self.secure_rx and self.secure_tx:
|
||||
rsp=self.app(cmd)
|
||||
else:
|
||||
rsp=self.usb.cmd(cmd)
|
||||
assert_status(rsp)
|
||||
sz, = unpack('<xxLxx', rsp[:8])
|
||||
|
||||
return rsp[8:8+sz]
|
||||
|
||||
def open(self):
|
||||
self.secure_rx = False
|
||||
self.secure_tx = False
|
||||
|
||||
self.parseTlsFlash(self.read_flash(1, 0, 0x1000))
|
||||
|
||||
self.handshake_hash = sha256()
|
||||
|
||||
rsp=self.usb.cmd(unhexlify('44000000') + self.make_handshake(self.make_client_hello()))
|
||||
|
|
@ -363,12 +377,7 @@ class Tls():
|
|||
def make_ext(self, id, b):
|
||||
return pack('>H', id) + with_2bytes_size(b)
|
||||
|
||||
def parse40(self, reply):
|
||||
h, reply = reply[:8], reply[8:]
|
||||
|
||||
if h != unhexlify('0000001000000000'):
|
||||
raise Exception('unexpected header: %s' % hexlify(h).decode())
|
||||
|
||||
def parseTlsFlash(self, reply):
|
||||
while len(reply) > 0:
|
||||
hdr, reply = reply[:4], reply[4:]
|
||||
hs, reply = reply[:0x20], reply[0x20:]
|
||||
|
|
@ -378,6 +387,12 @@ class Tls():
|
|||
if id == 0xffff:
|
||||
break
|
||||
|
||||
m=sha256()
|
||||
m.update(body)
|
||||
|
||||
if m.digest() != hs:
|
||||
raise Exception('hash mismatch')
|
||||
|
||||
if id == 4:
|
||||
self.handle_priv(body)
|
||||
elif id == 6:
|
||||
|
|
@ -391,12 +406,6 @@ class Tls():
|
|||
else:
|
||||
self.trace('unhandled block id %04x (%d bytes): %s' % (id, sz, hexlify(body)))
|
||||
|
||||
m=sha256()
|
||||
m.update(body)
|
||||
|
||||
if m.digest() != hs:
|
||||
raise Exception('hash mismatch')
|
||||
|
||||
def handle_empty(self, body):
|
||||
if body != b'\0' * len(body):
|
||||
raise Exception('Expected empty block')
|
||||
|
|
@ -404,7 +413,7 @@ class Tls():
|
|||
def handle_cert(self, body):
|
||||
# TODO validate cert, check if pub keys match
|
||||
self.tls_cert = body
|
||||
self.trace('TLS cert pub key: %s' % hexlify(self.tls_cert))
|
||||
self.trace('TLS cert blob: %s' % hexlify(self.tls_cert))
|
||||
|
||||
def handle_ecdh(self, body):
|
||||
# TODO check the signature
|
||||
|
|
|
|||
Loading…
Reference in a new issue