mirror of
https://github.com/brmlab/brmdoor_libnfc.git
synced 2025-11-02 03:23:58 +01:00
Change tabs to 4 spaces in python files
This commit is contained in:
parent
5d41737f80
commit
2482462a73
6 changed files with 420 additions and 420 deletions
|
|
@ -12,77 +12,77 @@ from optparse import OptionParser
|
|||
from brmdoor_nfc_daemon import BrmdoorConfig
|
||||
|
||||
def addUidAuth(cursor, uid_hex, nick):
|
||||
"""
|
||||
Add user authenticated by UID. UID should be in hex, 4, 7 or 10 bytes long.
|
||||
"""
|
||||
try:
|
||||
uid_hex.decode("hex")
|
||||
sql = """INSERT INTO authorized_uids
|
||||
(uid_hex, nick)
|
||||
values (?, ?)
|
||||
"""
|
||||
sql_data = (uid_hex, nick)
|
||||
cursor.execute(sql, sql_data)
|
||||
except TypeError:
|
||||
print >> sys.stderr, "UID must be in proper hex encoding"
|
||||
sys.exit(1)
|
||||
|
||||
"""
|
||||
Add user authenticated by UID. UID should be in hex, 4, 7 or 10 bytes long.
|
||||
"""
|
||||
try:
|
||||
uid_hex.decode("hex")
|
||||
sql = """INSERT INTO authorized_uids
|
||||
(uid_hex, nick)
|
||||
values (?, ?)
|
||||
"""
|
||||
sql_data = (uid_hex, nick)
|
||||
cursor.execute(sql, sql_data)
|
||||
except TypeError:
|
||||
print >> sys.stderr, "UID must be in proper hex encoding"
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def addHmacAuth(cursor, uid_hex, nick, key_hex):
|
||||
"""
|
||||
Add user authenticated by Yubikey HMAC-SHA1. UID should be in hex, 4, 7
|
||||
or 10 bytes long. HMAC key in key_hex must be exactly 20 bytes in hex.
|
||||
"""
|
||||
try:
|
||||
uid_hex.decode("hex")
|
||||
if len(key_hex.decode("hex")) != 20:
|
||||
print >> sys.stderr, "Key must be exactly 20 bytes long!"
|
||||
sys.exit(1)
|
||||
sql = """INSERT INTO authorized_hmac_keys
|
||||
(uid_hex, nick, key_hex)
|
||||
VALUES (?, ?, ?)
|
||||
"""
|
||||
sql_data = (uid_hex, nick, key_hex)
|
||||
cursor.execute(sql, sql_data)
|
||||
except TypeError:
|
||||
print >> sys.stderr, "UID and key must be in proper hex encoding"
|
||||
sys.exit(1)
|
||||
"""
|
||||
Add user authenticated by Yubikey HMAC-SHA1. UID should be in hex, 4, 7
|
||||
or 10 bytes long. HMAC key in key_hex must be exactly 20 bytes in hex.
|
||||
"""
|
||||
try:
|
||||
uid_hex.decode("hex")
|
||||
if len(key_hex.decode("hex")) != 20:
|
||||
print >> sys.stderr, "Key must be exactly 20 bytes long!"
|
||||
sys.exit(1)
|
||||
sql = """INSERT INTO authorized_hmac_keys
|
||||
(uid_hex, nick, key_hex)
|
||||
VALUES (?, ?, ?)
|
||||
"""
|
||||
sql_data = (uid_hex, nick, key_hex)
|
||||
cursor.execute(sql, sql_data)
|
||||
except TypeError:
|
||||
print >> sys.stderr, "UID and key must be in proper hex encoding"
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = OptionParser()
|
||||
parser.add_option("-c", "--config", action="store", type="string", dest="config",
|
||||
help="Configuration file")
|
||||
parser.add_option("-a", "--authtype", action="store", type="string", dest="authtype",
|
||||
help="Authenthication type - uid or hmac")
|
||||
(opts, args) = parser.parse_args()
|
||||
parser = OptionParser()
|
||||
parser.add_option("-c", "--config", action="store", type="string", dest="config",
|
||||
help="Configuration file")
|
||||
parser.add_option("-a", "--authtype", action="store", type="string", dest="authtype",
|
||||
help="Authenthication type - uid or hmac")
|
||||
(opts, args) = parser.parse_args()
|
||||
|
||||
if opts.config is None:
|
||||
print >> sys.stderr, "You must specify config file via the -c option!"
|
||||
parser.print_help()
|
||||
sys.exit(1)
|
||||
|
||||
if opts.authtype not in ["uid", "hmac"]:
|
||||
print >> sys.stderr, "You must specify authentication type via -a option!"
|
||||
print >> sys.stderr, "Acceptable choices: uid, hmac"
|
||||
sys.exit(1)
|
||||
|
||||
config = BrmdoorConfig(opts.config)
|
||||
conn = sqlite3.connect(config.authDbFilename)
|
||||
cursor = conn.cursor()
|
||||
|
||||
if opts.authtype == "uid":
|
||||
if len(args) < 2:
|
||||
print >> sys.stderr, "You must two additional arguments, hex UID and nick"
|
||||
print >> sys.stderr, "Example:"
|
||||
print >> sys.stderr, "brmdoor_adduser.py -c brmdoor.config -a uid 34795FCC SomeUserName"
|
||||
sys.exit(1)
|
||||
addUidAuth(cursor, args[0], args[1])
|
||||
elif opts.authtype == "hmac":
|
||||
if len(args) < 3:
|
||||
print >> sys.stderr, "You must three additional arguments, hex UID and nick and hex key"
|
||||
print >> sys.stderr, "brmdoor_adduser.py -c brmdoor.config -a hmac 40795FCCAB0701 SomeUserName 000102030405060708090a0b0c0d0e0f31323334"
|
||||
sys.exit(1)
|
||||
addHmacAuth(cursor, args[0], args[1], args[2])
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
if opts.config is None:
|
||||
print >> sys.stderr, "You must specify config file via the -c option!"
|
||||
parser.print_help()
|
||||
sys.exit(1)
|
||||
|
||||
if opts.authtype not in ["uid", "hmac"]:
|
||||
print >> sys.stderr, "You must specify authentication type via -a option!"
|
||||
print >> sys.stderr, "Acceptable choices: uid, hmac"
|
||||
sys.exit(1)
|
||||
|
||||
config = BrmdoorConfig(opts.config)
|
||||
conn = sqlite3.connect(config.authDbFilename)
|
||||
cursor = conn.cursor()
|
||||
|
||||
if opts.authtype == "uid":
|
||||
if len(args) < 2:
|
||||
print >> sys.stderr, "You must two additional arguments, hex UID and nick"
|
||||
print >> sys.stderr, "Example:"
|
||||
print >> sys.stderr, "brmdoor_adduser.py -c brmdoor.config -a uid 34795FCC SomeUserName"
|
||||
sys.exit(1)
|
||||
addUidAuth(cursor, args[0], args[1])
|
||||
elif opts.authtype == "hmac":
|
||||
if len(args) < 3:
|
||||
print >> sys.stderr, "You must three additional arguments, hex UID and nick and hex key"
|
||||
print >> sys.stderr, "brmdoor_adduser.py -c brmdoor.config -a hmac 40795FCCAB0701 SomeUserName 000102030405060708090a0b0c0d0e0f31323334"
|
||||
sys.exit(1)
|
||||
addHmacAuth(cursor, args[0], args[1], args[2])
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
|
|
|||
|
|
@ -8,150 +8,150 @@ from nfc_smartcard import NFCError
|
|||
|
||||
|
||||
class UidRecord(object):
|
||||
"""Represents UID<->nick pair"""
|
||||
|
||||
def __init__(self, uid_hex, nick):
|
||||
"""
|
||||
Create instance binding UID to nick. UIDs should be either 4, 7
|
||||
or 10 bytes long, but that's ISO14443 thing - this object has
|
||||
no such limitation.
|
||||
|
||||
UID will be stored in uppercase hex, converted if necessary.
|
||||
|
||||
@param uid_hex: uid in hex
|
||||
@param nick: nickname this UID belongs to
|
||||
"""
|
||||
self.uid_hex = uid_hex.upper()
|
||||
self.nick = nick
|
||||
|
||||
def __str__(self):
|
||||
return "(uid: %s, nick: %s)" % (self.uid_hex, self.nick)
|
||||
|
||||
def __repr__(self):
|
||||
return "<UidRecord: uid: %s, nick: %s>" % \
|
||||
(repr(self.uid_hex), repr(self.nick))
|
||||
"""Represents UID<->nick pair"""
|
||||
|
||||
def __init__(self, uid_hex, nick):
|
||||
"""
|
||||
Create instance binding UID to nick. UIDs should be either 4, 7
|
||||
or 10 bytes long, but that's ISO14443 thing - this object has
|
||||
no such limitation.
|
||||
|
||||
UID will be stored in uppercase hex, converted if necessary.
|
||||
|
||||
@param uid_hex: uid in hex
|
||||
@param nick: nickname this UID belongs to
|
||||
"""
|
||||
self.uid_hex = uid_hex.upper()
|
||||
self.nick = nick
|
||||
|
||||
def __str__(self):
|
||||
return "(uid: %s, nick: %s)" % (self.uid_hex, self.nick)
|
||||
|
||||
def __repr__(self):
|
||||
return "<UidRecord: uid: %s, nick: %s>" % \
|
||||
(repr(self.uid_hex), repr(self.nick))
|
||||
|
||||
class UidAuthenticator(object):
|
||||
"""Checks UIDs of ISO14443 RFID cards against database."""
|
||||
|
||||
def __init__(self, filename):
|
||||
"""
|
||||
Connects to database by given filename and later checks UIDs
|
||||
against that database.
|
||||
"""
|
||||
#open in autocommit mode - we are not changing anything
|
||||
self.conn = sqlite3.connect(filename, isolation_level=None)
|
||||
"""Checks UIDs of ISO14443 RFID cards against database."""
|
||||
|
||||
def __init__(self, filename):
|
||||
"""
|
||||
Connects to database by given filename and later checks UIDs
|
||||
against that database.
|
||||
"""
|
||||
#open in autocommit mode - we are not changing anything
|
||||
self.conn = sqlite3.connect(filename, isolation_level=None)
|
||||
|
||||
def fetchUidRecord(self, uid_hex):
|
||||
"""
|
||||
Returns first record that matches given UID or None if nothing
|
||||
is found.
|
||||
|
||||
@param uid_hex: uid to match in hex
|
||||
@returns UidRecord instance if found, None otherwise
|
||||
"""
|
||||
cursor = self.conn.cursor()
|
||||
sql = "SELECT nick FROM authorized_uids WHERE UPPER(uid_hex)=?"
|
||||
sql_data =(uid_hex.upper(),)
|
||||
|
||||
cursor.execute(sql, sql_data)
|
||||
record = cursor.fetchone()
|
||||
|
||||
if record is None:
|
||||
return None
|
||||
|
||||
nick = record[0]
|
||||
return UidRecord(uid_hex, nick)
|
||||
|
||||
def shutdown(self):
|
||||
"""Closes connection to database"""
|
||||
self.conn.close()
|
||||
def fetchUidRecord(self, uid_hex):
|
||||
"""
|
||||
Returns first record that matches given UID or None if nothing
|
||||
is found.
|
||||
|
||||
@param uid_hex: uid to match in hex
|
||||
@returns UidRecord instance if found, None otherwise
|
||||
"""
|
||||
cursor = self.conn.cursor()
|
||||
sql = "SELECT nick FROM authorized_uids WHERE UPPER(uid_hex)=?"
|
||||
sql_data =(uid_hex.upper(),)
|
||||
|
||||
cursor.execute(sql, sql_data)
|
||||
record = cursor.fetchone()
|
||||
|
||||
if record is None:
|
||||
return None
|
||||
|
||||
nick = record[0]
|
||||
return UidRecord(uid_hex, nick)
|
||||
|
||||
def shutdown(self):
|
||||
"""Closes connection to database"""
|
||||
self.conn.close()
|
||||
|
||||
|
||||
class YubikeyHMACAuthenthicator(object):
|
||||
"""
|
||||
Uses Yubikey Neo's built-in HMAC functionality on slot 2 (needs to be
|
||||
configured using Yubikey tools to be on this slot).
|
||||
"""
|
||||
def __init__(self, filename, nfcReader):
|
||||
"""
|
||||
Connects to database by given filename and later checks UIDs
|
||||
against that database.
|
||||
"""
|
||||
#again autocommit mode
|
||||
self.conn = sqlite3.connect(filename, isolation_level=None)
|
||||
self.nfcReader = nfcReader
|
||||
|
||||
def hmacCheck(self, key, challenge, result):
|
||||
"""
|
||||
Returns true iff HMAC-SHA1 with given key and challenge string
|
||||
transforms into given result.
|
||||
"""
|
||||
hashed = hmac.new(key, challenge, hashlib.sha1)
|
||||
#We should use hmac.compare_digest(), but that's in new Python
|
||||
#version only. Here timing side channels are not much of concern.
|
||||
return hashed.digest() == result
|
||||
"""
|
||||
Uses Yubikey Neo's built-in HMAC functionality on slot 2 (needs to be
|
||||
configured using Yubikey tools to be on this slot).
|
||||
"""
|
||||
def __init__(self, filename, nfcReader):
|
||||
"""
|
||||
Connects to database by given filename and later checks UIDs
|
||||
against that database.
|
||||
"""
|
||||
#again autocommit mode
|
||||
self.conn = sqlite3.connect(filename, isolation_level=None)
|
||||
self.nfcReader = nfcReader
|
||||
|
||||
def hmacCheck(self, key, challenge, result):
|
||||
"""
|
||||
Returns true iff HMAC-SHA1 with given key and challenge string
|
||||
transforms into given result.
|
||||
"""
|
||||
hashed = hmac.new(key, challenge, hashlib.sha1)
|
||||
#We should use hmac.compare_digest(), but that's in new Python
|
||||
#version only. Here timing side channels are not much of concern.
|
||||
return hashed.digest() == result
|
||||
|
||||
def checkHMACforUID(self, uid_hex):
|
||||
"""
|
||||
Checks if UID is in database. If so
|
||||
@param uid_hex: uid to match in hex
|
||||
@returns UidRecord instance if found, None otherwise
|
||||
"""
|
||||
cursor = self.conn.cursor()
|
||||
sql = "SELECT nick, key_hex FROM authorized_hmac_keys WHERE UPPER(uid_hex)=?"
|
||||
sql_data =(uid_hex.upper(),)
|
||||
|
||||
cursor.execute(sql, sql_data)
|
||||
record = cursor.fetchone()
|
||||
|
||||
if record is None:
|
||||
return None
|
||||
|
||||
nick = record[0]
|
||||
secretKey = record[1].decode("hex")
|
||||
|
||||
challenge = os.urandom(32)
|
||||
|
||||
# Select HMAC-SHA1 on slot 2 from Yubikey
|
||||
apdusHex = [
|
||||
"00 A4 04 00 07 A0 00 00 05 27 20 01",
|
||||
"00 01 38 00 %02x %s" % (len(challenge), challenge.encode("hex"))
|
||||
]
|
||||
|
||||
rapdu = None
|
||||
|
||||
for apduHex in apdusHex:
|
||||
try:
|
||||
apdu = apduHex.replace(" ", "").decode("hex")
|
||||
rapdu = self.nfcReader.sendAPDU(apdu)
|
||||
if not rapdu.valid or rapdu.sw() != 0x9000:
|
||||
raise NFCError("HMAC - response SW is not 0x9000")
|
||||
except NFCError, e:
|
||||
logging.debug("Yubikey HMAC command failed: %s" % e.what())
|
||||
return None
|
||||
|
||||
if not self.hmacCheck(secretKey, challenge, rapdu.data()):
|
||||
logging.info("HMAC check failed for UID %s", uid_hex)
|
||||
return None
|
||||
|
||||
return UidRecord(uid_hex, nick)
|
||||
|
||||
def shutdown(self):
|
||||
"""Closes connection to database"""
|
||||
self.conn.close()
|
||||
|
||||
def checkHMACforUID(self, uid_hex):
|
||||
"""
|
||||
Checks if UID is in database. If so
|
||||
@param uid_hex: uid to match in hex
|
||||
@returns UidRecord instance if found, None otherwise
|
||||
"""
|
||||
cursor = self.conn.cursor()
|
||||
sql = "SELECT nick, key_hex FROM authorized_hmac_keys WHERE UPPER(uid_hex)=?"
|
||||
sql_data =(uid_hex.upper(),)
|
||||
|
||||
cursor.execute(sql, sql_data)
|
||||
record = cursor.fetchone()
|
||||
|
||||
if record is None:
|
||||
return None
|
||||
|
||||
nick = record[0]
|
||||
secretKey = record[1].decode("hex")
|
||||
|
||||
challenge = os.urandom(32)
|
||||
|
||||
# Select HMAC-SHA1 on slot 2 from Yubikey
|
||||
apdusHex = [
|
||||
"00 A4 04 00 07 A0 00 00 05 27 20 01",
|
||||
"00 01 38 00 %02x %s" % (len(challenge), challenge.encode("hex"))
|
||||
]
|
||||
|
||||
rapdu = None
|
||||
|
||||
for apduHex in apdusHex:
|
||||
try:
|
||||
apdu = apduHex.replace(" ", "").decode("hex")
|
||||
rapdu = self.nfcReader.sendAPDU(apdu)
|
||||
if not rapdu.valid or rapdu.sw() != 0x9000:
|
||||
raise NFCError("HMAC - response SW is not 0x9000")
|
||||
except NFCError, e:
|
||||
logging.debug("Yubikey HMAC command failed: %s" % e.what())
|
||||
return None
|
||||
|
||||
if not self.hmacCheck(secretKey, challenge, rapdu.data()):
|
||||
logging.info("HMAC check failed for UID %s", uid_hex)
|
||||
return None
|
||||
|
||||
return UidRecord(uid_hex, nick)
|
||||
|
||||
def shutdown(self):
|
||||
"""Closes connection to database"""
|
||||
self.conn.close()
|
||||
|
||||
#test routine
|
||||
if __name__ == "__main__":
|
||||
authenticator = UidAuthenticator("test_uids_db.sqlite")
|
||||
|
||||
record = authenticator.fetchUidRecord("043a1482cc2280")
|
||||
print "For UID 043a1482cc2280 we found:", repr(record)
|
||||
|
||||
record = authenticator.fetchUidRecord("34795fad")
|
||||
print "For UID 34795fad we found:", repr(record)
|
||||
|
||||
record = authenticator.fetchUidRecord("01020304")
|
||||
print "For UID 01020304 we found:", repr(record)
|
||||
|
||||
authenticator.shutdown()
|
||||
authenticator = UidAuthenticator("test_uids_db.sqlite")
|
||||
|
||||
record = authenticator.fetchUidRecord("043a1482cc2280")
|
||||
print "For UID 043a1482cc2280 we found:", repr(record)
|
||||
|
||||
record = authenticator.fetchUidRecord("34795fad")
|
||||
print "For UID 34795fad we found:", repr(record)
|
||||
|
||||
record = authenticator.fetchUidRecord("01020304")
|
||||
print "For UID 01020304 we found:", repr(record)
|
||||
|
||||
authenticator.shutdown()
|
||||
|
|
|
|||
|
|
@ -13,138 +13,138 @@ from brmdoor_authenticator import UidAuthenticator, YubikeyHMACAuthenthicator
|
|||
import unlocker
|
||||
|
||||
class BrmdoorConfigError(ConfigParser.Error):
|
||||
"""
|
||||
Signifies that config has missing or bad values.
|
||||
"""
|
||||
pass
|
||||
"""
|
||||
Signifies that config has missing or bad values.
|
||||
"""
|
||||
pass
|
||||
|
||||
class BrmdoorConfig(object):
|
||||
"""
|
||||
Configuration parser. Holds config variables from config file.
|
||||
"""
|
||||
|
||||
_defaults = {
|
||||
"lock_opened_secs": "5",
|
||||
"unknown_uid_timeout_secs": "5",
|
||||
"log_level": "info"
|
||||
}
|
||||
|
||||
def __init__(self, filename):
|
||||
"""
|
||||
Parse and read config from given filename.
|
||||
|
||||
@throws ConfigParser.Error if parsing failed
|
||||
@throws BrmdoorConfigError if some value was missing or invalid
|
||||
"""
|
||||
self.config = ConfigParser.SafeConfigParser(defaults=BrmdoorConfig._defaults)
|
||||
self.config.read(filename)
|
||||
|
||||
self.authDbFilename = self.config.get("brmdoor", "auth_db_filename")
|
||||
self.lockOpenedSecs = self.config.getint("brmdoor", "lock_opened_secs")
|
||||
self.unknownUidTimeoutSecs = self.config.getint("brmdoor", "unknown_uid_timeout_secs")
|
||||
self.logFile = self.config.get("brmdoor", "log_file")
|
||||
self.logLevel = self.convertLoglevel(self.config.get("brmdoor", "log_level"))
|
||||
self.unlocker = self.config.get("brmdoor", "unlocker")
|
||||
|
||||
def convertLoglevel(self, levelString):
|
||||
"""Converts string 'debug', 'info', etc. into corresponding
|
||||
logging.XXX value which is returned.
|
||||
|
||||
@raises BrmdoorConfigError if the level is undefined
|
||||
"""
|
||||
try:
|
||||
return getattr(logging, levelString.upper())
|
||||
except AttributeError:
|
||||
raise BrmdoorConfigError("No such loglevel - %s" % levelString)
|
||||
"""
|
||||
Configuration parser. Holds config variables from config file.
|
||||
"""
|
||||
|
||||
_defaults = {
|
||||
"lock_opened_secs": "5",
|
||||
"unknown_uid_timeout_secs": "5",
|
||||
"log_level": "info"
|
||||
}
|
||||
|
||||
def __init__(self, filename):
|
||||
"""
|
||||
Parse and read config from given filename.
|
||||
|
||||
@throws ConfigParser.Error if parsing failed
|
||||
@throws BrmdoorConfigError if some value was missing or invalid
|
||||
"""
|
||||
self.config = ConfigParser.SafeConfigParser(defaults=BrmdoorConfig._defaults)
|
||||
self.config.read(filename)
|
||||
|
||||
self.authDbFilename = self.config.get("brmdoor", "auth_db_filename")
|
||||
self.lockOpenedSecs = self.config.getint("brmdoor", "lock_opened_secs")
|
||||
self.unknownUidTimeoutSecs = self.config.getint("brmdoor", "unknown_uid_timeout_secs")
|
||||
self.logFile = self.config.get("brmdoor", "log_file")
|
||||
self.logLevel = self.convertLoglevel(self.config.get("brmdoor", "log_level"))
|
||||
self.unlocker = self.config.get("brmdoor", "unlocker")
|
||||
|
||||
def convertLoglevel(self, levelString):
|
||||
"""Converts string 'debug', 'info', etc. into corresponding
|
||||
logging.XXX value which is returned.
|
||||
|
||||
@raises BrmdoorConfigError if the level is undefined
|
||||
"""
|
||||
try:
|
||||
return getattr(logging, levelString.upper())
|
||||
except AttributeError:
|
||||
raise BrmdoorConfigError("No such loglevel - %s" % levelString)
|
||||
|
||||
class NFCScanner(object):
|
||||
"""Thread reading data from NFC reader"""
|
||||
|
||||
def __init__(self, config):
|
||||
"""Create worker reading UIDs from PN53x reader.
|
||||
"""
|
||||
self.authenticator = UidAuthenticator(config.authDbFilename)
|
||||
self.hmacAuthenticator = None
|
||||
self.unknownUidTimeoutSecs = config.unknownUidTimeoutSecs
|
||||
self.lockOpenedSecs = config.lockOpenedSecs
|
||||
|
||||
unlockerClassName = config.unlocker
|
||||
unlockerClass = getattr(unlocker, unlockerClassName)
|
||||
self.unlocker = unlockerClass(config)
|
||||
"""Thread reading data from NFC reader"""
|
||||
|
||||
def __init__(self, config):
|
||||
"""Create worker reading UIDs from PN53x reader.
|
||||
"""
|
||||
self.authenticator = UidAuthenticator(config.authDbFilename)
|
||||
self.hmacAuthenticator = None
|
||||
self.unknownUidTimeoutSecs = config.unknownUidTimeoutSecs
|
||||
self.lockOpenedSecs = config.lockOpenedSecs
|
||||
|
||||
unlockerClassName = config.unlocker
|
||||
unlockerClass = getattr(unlocker, unlockerClassName)
|
||||
self.unlocker = unlockerClass(config)
|
||||
|
||||
def run(self):
|
||||
"""
|
||||
Waits for a card to get into reader field. Reads its UID and
|
||||
compares to database of authorized users. Unlocks lock if
|
||||
authorized.
|
||||
"""
|
||||
self.nfc = NFCDevice()
|
||||
self.hmacAuthenticator = YubikeyHMACAuthenthicator(
|
||||
config.authDbFilename, self.nfc
|
||||
)
|
||||
#self.nfc.pollNr = 0xFF #poll indefinitely
|
||||
while True:
|
||||
try:
|
||||
uid_hex = hexlify(self.nfc.scanUID())
|
||||
logging.debug("Got UID %s", uid_hex)
|
||||
if len(uid_hex) > 0:
|
||||
self.actOnUid(uid_hex)
|
||||
else:
|
||||
#prevent busy loop if reader goes awry
|
||||
time.sleep(0.3)
|
||||
except NFCError, e:
|
||||
#this exception happens also when scanUID times out
|
||||
logging.debug("Failed to wait for RFID card: %s", e)
|
||||
except KeyboardInterrupt:
|
||||
logging.info("Exiting on keyboard interrupt")
|
||||
self.nfc.close()
|
||||
self.nfc.unload()
|
||||
self.unlocker.lock()
|
||||
sys.exit(2)
|
||||
except Exception:
|
||||
logging.exception("Exception in main unlock thread")
|
||||
|
||||
def actOnUid(self, uid_hex):
|
||||
"""
|
||||
Do something with the UID scanned. Try to authenticate it against
|
||||
database and open lock if authorized.
|
||||
"""
|
||||
record = self.authenticator.fetchUidRecord(uid_hex)
|
||||
|
||||
#direct UID match
|
||||
if record is not None:
|
||||
logging.info("Unlocking for UID %s", record)
|
||||
self.unlocker.unlock()
|
||||
return
|
||||
|
||||
#test for Yubikey HMAC auth
|
||||
record = self.hmacAuthenticator.checkHMACforUID(uid_hex)
|
||||
|
||||
if record is not None:
|
||||
logging.info("Unlocking after HMAC for UID %s", record)
|
||||
self.unlocker.unlock()
|
||||
return
|
||||
|
||||
logging.info("Unknown UID %s", uid_hex)
|
||||
time.sleep(self.unknownUidTimeoutSecs)
|
||||
|
||||
def run(self):
|
||||
"""
|
||||
Waits for a card to get into reader field. Reads its UID and
|
||||
compares to database of authorized users. Unlocks lock if
|
||||
authorized.
|
||||
"""
|
||||
self.nfc = NFCDevice()
|
||||
self.hmacAuthenticator = YubikeyHMACAuthenthicator(
|
||||
config.authDbFilename, self.nfc
|
||||
)
|
||||
#self.nfc.pollNr = 0xFF #poll indefinitely
|
||||
while True:
|
||||
try:
|
||||
uid_hex = hexlify(self.nfc.scanUID())
|
||||
logging.debug("Got UID %s", uid_hex)
|
||||
if len(uid_hex) > 0:
|
||||
self.actOnUid(uid_hex)
|
||||
else:
|
||||
#prevent busy loop if reader goes awry
|
||||
time.sleep(0.3)
|
||||
except NFCError, e:
|
||||
#this exception happens also when scanUID times out
|
||||
logging.debug("Failed to wait for RFID card: %s", e)
|
||||
except KeyboardInterrupt:
|
||||
logging.info("Exiting on keyboard interrupt")
|
||||
self.nfc.close()
|
||||
self.nfc.unload()
|
||||
self.unlocker.lock()
|
||||
sys.exit(2)
|
||||
except Exception:
|
||||
logging.exception("Exception in main unlock thread")
|
||||
|
||||
def actOnUid(self, uid_hex):
|
||||
"""
|
||||
Do something with the UID scanned. Try to authenticate it against
|
||||
database and open lock if authorized.
|
||||
"""
|
||||
record = self.authenticator.fetchUidRecord(uid_hex)
|
||||
|
||||
#direct UID match
|
||||
if record is not None:
|
||||
logging.info("Unlocking for UID %s", record)
|
||||
self.unlocker.unlock()
|
||||
return
|
||||
|
||||
#test for Yubikey HMAC auth
|
||||
record = self.hmacAuthenticator.checkHMACforUID(uid_hex)
|
||||
|
||||
if record is not None:
|
||||
logging.info("Unlocking after HMAC for UID %s", record)
|
||||
self.unlocker.unlock()
|
||||
return
|
||||
|
||||
logging.info("Unknown UID %s", uid_hex)
|
||||
time.sleep(self.unknownUidTimeoutSecs)
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
if len(sys.argv) < 2:
|
||||
print >> sys.stderr, "Syntax: brmdoor_nfc_daemon.py brmdoor_nfc.config"
|
||||
sys.exit(1)
|
||||
|
||||
config = BrmdoorConfig(sys.argv[1])
|
||||
|
||||
if config.logFile == "-":
|
||||
logging.basicConfig(stream=sys.stderr, level=config.logLevel,
|
||||
format="%(asctime)s %(levelname)s %(message)s [%(pathname)s:%(lineno)d]")
|
||||
else:
|
||||
logging.basicConfig(filename=config.logFile, level=config.logLevel,
|
||||
format="%(asctime)s %(levelname)s %(message)s [%(pathname)s:%(lineno)d]")
|
||||
|
||||
nfcScanner = NFCScanner(config)
|
||||
nfcScanner.run()
|
||||
|
||||
|
||||
if len(sys.argv) < 2:
|
||||
print >> sys.stderr, "Syntax: brmdoor_nfc_daemon.py brmdoor_nfc.config"
|
||||
sys.exit(1)
|
||||
|
||||
config = BrmdoorConfig(sys.argv[1])
|
||||
|
||||
if config.logFile == "-":
|
||||
logging.basicConfig(stream=sys.stderr, level=config.logLevel,
|
||||
format="%(asctime)s %(levelname)s %(message)s [%(pathname)s:%(lineno)d]")
|
||||
else:
|
||||
logging.basicConfig(filename=config.logFile, level=config.logLevel,
|
||||
format="%(asctime)s %(levelname)s %(message)s [%(pathname)s:%(lineno)d]")
|
||||
|
||||
nfcScanner = NFCScanner(config)
|
||||
nfcScanner.run()
|
||||
|
||||
|
|
|
|||
|
|
@ -10,23 +10,23 @@ import sys
|
|||
import sqlite3
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2:
|
||||
print >> sys.stderr, "You must specify filename as arg1 where the DB is to be created"
|
||||
|
||||
filename = sys.argv[1]
|
||||
conn = sqlite3.connect(filename)
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute("""CREATE TABLE authorized_uids(
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
uid_hex TEXT,
|
||||
nick TEXT)
|
||||
""")
|
||||
cursor.execute("""CREATE TABLE authorized_hmac_keys(
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
uid_hex TEXT,
|
||||
key_hex TEXT,
|
||||
nick TEXT)
|
||||
""")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
if len(sys.argv) < 2:
|
||||
print >> sys.stderr, "You must specify filename as arg1 where the DB is to be created"
|
||||
|
||||
filename = sys.argv[1]
|
||||
conn = sqlite3.connect(filename)
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute("""CREATE TABLE authorized_uids(
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
uid_hex TEXT,
|
||||
nick TEXT)
|
||||
""")
|
||||
cursor.execute("""CREATE TABLE authorized_hmac_keys(
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
uid_hex TEXT,
|
||||
key_hex TEXT,
|
||||
nick TEXT)
|
||||
""")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
|
|
|||
44
test_nfc.py
44
test_nfc.py
|
|
@ -5,8 +5,8 @@ from nfc_smartcard import NFCDevice, NFCError
|
|||
from binascii import hexlify
|
||||
|
||||
def formatAPDU(apdu):
|
||||
return " ".join(["%02X" % ord(b) for b in apdu])
|
||||
|
||||
return " ".join(["%02X" % ord(b) for b in apdu])
|
||||
|
||||
tests = {
|
||||
# Reading of file E104, where usually NDEF message is
|
||||
"ndef4": [
|
||||
|
|
@ -55,26 +55,26 @@ hex_apdus = tests[apdu_test]
|
|||
apdus = [hex_apdu.replace(" ","").decode("hex") for hex_apdu in hex_apdus]
|
||||
|
||||
try:
|
||||
nfc = NFCDevice()
|
||||
uid = nfc.scanUID()
|
||||
print "UID", hexlify(uid)
|
||||
#nfc.close()
|
||||
#nfc.open()
|
||||
nfc = NFCDevice()
|
||||
uid = nfc.scanUID()
|
||||
print "UID", hexlify(uid)
|
||||
#nfc.close()
|
||||
#nfc.open()
|
||||
|
||||
print "Now trying to send ISO14443-4 APDUs"
|
||||
try:
|
||||
#nfc.selectPassiveTarget()
|
||||
for apdu in apdus:
|
||||
print "Command APDU:", formatAPDU(apdu)
|
||||
rapdu = nfc.sendAPDU(apdu)
|
||||
print "Response APDU valid: %s, SW %04x, data %s" % (rapdu.valid(), rapdu.sw(), hexlify(rapdu.data()))
|
||||
except NFCError, e:
|
||||
print "Failed to transmit APDU:", e.what()
|
||||
print "Now trying to send ISO14443-4 APDUs"
|
||||
try:
|
||||
#nfc.selectPassiveTarget()
|
||||
for apdu in apdus:
|
||||
print "Command APDU:", formatAPDU(apdu)
|
||||
rapdu = nfc.sendAPDU(apdu)
|
||||
print "Response APDU valid: %s, SW %04x, data %s" % (rapdu.valid(), rapdu.sw(), hexlify(rapdu.data()))
|
||||
except NFCError, e:
|
||||
print "Failed to transmit APDU:", e.what()
|
||||
|
||||
print "Device is opened:", nfc.opened()
|
||||
print "Closing device"
|
||||
nfc.close()
|
||||
print "Device is opened:", nfc.opened()
|
||||
nfc.unload()
|
||||
print "Device is opened:", nfc.opened()
|
||||
print "Closing device"
|
||||
nfc.close()
|
||||
print "Device is opened:", nfc.opened()
|
||||
nfc.unload()
|
||||
except NFCError, e:
|
||||
print "Reading UID failed:", e.what()
|
||||
print "Reading UID failed:", e.what()
|
||||
|
|
|
|||
88
unlocker.py
88
unlocker.py
|
|
@ -2,55 +2,55 @@ import time
|
|||
import wiringpi2 as wiringpi
|
||||
|
||||
class Unlocker(object):
|
||||
"""Abstract class/interface for Unlocker object.
|
||||
Unlocker useful for simulation, but does not actually unlock any lock.
|
||||
"""
|
||||
def __init__(self, config):
|
||||
"""
|
||||
Creates unlocked instance from config, where section named after
|
||||
the class is supposed to exist.
|
||||
|
||||
@param config: BrmdoorConfig instance
|
||||
"""
|
||||
self.config = config.config
|
||||
self.lockOpenedSecs = config.lockOpenedSecs
|
||||
self.unlockerName = type(self).__name__
|
||||
|
||||
def unlock(self):
|
||||
"""Unlock lock for given self.lockOpenedSecs.
|
||||
In this class case, it's only simulated
|
||||
"""
|
||||
time.sleep(self.lockOpenedSecs)
|
||||
"""Abstract class/interface for Unlocker object.
|
||||
Unlocker useful for simulation, but does not actually unlock any lock.
|
||||
"""
|
||||
def __init__(self, config):
|
||||
"""
|
||||
Creates unlocked instance from config, where section named after
|
||||
the class is supposed to exist.
|
||||
|
||||
@param config: BrmdoorConfig instance
|
||||
"""
|
||||
self.config = config.config
|
||||
self.lockOpenedSecs = config.lockOpenedSecs
|
||||
self.unlockerName = type(self).__name__
|
||||
|
||||
def unlock(self):
|
||||
"""Unlock lock for given self.lockOpenedSecs.
|
||||
In this class case, it's only simulated
|
||||
"""
|
||||
time.sleep(self.lockOpenedSecs)
|
||||
|
||||
def lock(self):
|
||||
"""
|
||||
Lock the lock back. Meant to be used when program is shut down
|
||||
so that lock is not left disengaged.
|
||||
"""
|
||||
pass
|
||||
def lock(self):
|
||||
"""
|
||||
Lock the lock back. Meant to be used when program is shut down
|
||||
so that lock is not left disengaged.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class UnlockerWiringPi(Unlocker):
|
||||
"""Uses configured pings via WiringPi to open lock.
|
||||
"""
|
||||
"""Uses configured pings via WiringPi to open lock.
|
||||
"""
|
||||
|
||||
def __init__(self, config):
|
||||
Unlocker.__init__(self, config)
|
||||
def __init__(self, config):
|
||||
Unlocker.__init__(self, config)
|
||||
wiringpi.wiringPiSetupGpio() # pin numbers follow P1 GPIO header
|
||||
self.lockPin = self.config.getint("UnlockerWiringPi", "lock_pin")
|
||||
wiringpi.pinMode(self.lockPin, 1) #output
|
||||
|
||||
def unlock(self):
|
||||
"""Unlocks lock at configured pin by pulling it high.
|
||||
"""
|
||||
wiringpi.digitalWrite(self.lockPin, 1)
|
||||
time.sleep(self.lockOpenedSecs)
|
||||
wiringpi.digitalWrite(self.lockPin, 0)
|
||||
self.lockPin = self.config.getint("UnlockerWiringPi", "lock_pin")
|
||||
wiringpi.pinMode(self.lockPin, 1) #output
|
||||
|
||||
def unlock(self):
|
||||
"""Unlocks lock at configured pin by pulling it high.
|
||||
"""
|
||||
wiringpi.digitalWrite(self.lockPin, 1)
|
||||
time.sleep(self.lockOpenedSecs)
|
||||
wiringpi.digitalWrite(self.lockPin, 0)
|
||||
|
||||
def lock(self):
|
||||
"""
|
||||
Lock the lock back. Meant to be used when program is shut down
|
||||
so that lock is not left disengaged.
|
||||
"""
|
||||
wiringpi.digitalWrite(self.lockPin, 0)
|
||||
def lock(self):
|
||||
"""
|
||||
Lock the lock back. Meant to be used when program is shut down
|
||||
so that lock is not left disengaged.
|
||||
"""
|
||||
wiringpi.digitalWrite(self.lockPin, 0)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue