343 lines
10 KiB
Python
343 lines
10 KiB
Python
# Copyright (c) 2018 Yubico AB
|
|
# All rights reserved.
|
|
#
|
|
# Redistribution and use in source and binary forms, with or
|
|
# without modification, are permitted provided that the following
|
|
# conditions are met:
|
|
#
|
|
# 1. Redistributions of source code must retain the above copyright
|
|
# notice, this list of conditions and the following disclaimer.
|
|
# 2. Redistributions in binary form must reproduce the above
|
|
# copyright notice, this list of conditions and the following
|
|
# disclaimer in the documentation and/or other materials provided
|
|
# with the distribution.
|
|
#
|
|
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
|
|
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
|
|
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
|
|
# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
|
|
# COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
|
|
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
|
|
# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
|
|
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
|
|
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
|
|
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
|
|
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
|
|
# POSSIBILITY OF SUCH DAMAGE.
|
|
|
|
"""
|
|
Connects to the first FIDO device found (starts from USB, then looks into NFC),
|
|
creates a new credential for it, and authenticates the credential.
|
|
This works with both FIDO 2.0 devices as well as with U2F devices.
|
|
On Windows, the native WebAuthn API will be used.
|
|
"""
|
|
import time
|
|
import configparser
|
|
gpie = True
|
|
if gpie:
|
|
import RPi.GPIO as GPIO
|
|
import pickle
|
|
import psycopg2
|
|
import fido2.webauthn
|
|
from fido2.hid import CtapHidDevice
|
|
from fido2.client import Fido2Client, WindowsClient, UserInteraction
|
|
from fido2.server import Fido2Server
|
|
from getpass import getpass
|
|
import sys
|
|
import ctypes
|
|
from fido2.ctap2 import Ctap2
|
|
from fido2 import cbor
|
|
# from fido2 import CtapNfcDevice
|
|
|
|
config = configparser.ConfigParser()
|
|
|
|
config.read('config.ini')
|
|
# test
|
|
dbusername = config.get('Setup', 'dbusername')
|
|
dbpassword = config.get('Setup', 'dbpassword')
|
|
dbname = config.get('Setup', 'dbname')
|
|
dbhost = config.get('Setup', 'dbhost')
|
|
doorID = config.get('Setup', 'doorID')
|
|
|
|
gotAaguid = None
|
|
if gpie:
|
|
GPIO.setmode(GPIO.BOARD)
|
|
|
|
secLevel = 0
|
|
user_id = None
|
|
greenLed = 8
|
|
redLed = 7
|
|
|
|
door = 40
|
|
|
|
if gpie:
|
|
# badCredentials = False
|
|
GPIO.setup(greenLed, GPIO.OUT)
|
|
GPIO.output(greenLed, GPIO.LOW)
|
|
GPIO.setup(redLed, GPIO.OUT)
|
|
|
|
GPIO.output(redLed, GPIO.HIGH)
|
|
time.sleep(1)
|
|
GPIO.output(redLed, GPIO.LOW)
|
|
time.sleep(0.2)
|
|
|
|
GPIO.output(greenLed, GPIO.HIGH)
|
|
time.sleep(1)
|
|
GPIO.output(greenLed, GPIO.LOW)
|
|
time.sleep(0.2)
|
|
|
|
GPIO.setup(door, GPIO.OUT)
|
|
GPIO.output(door, GPIO.LOW)
|
|
def badCred():
|
|
print("Bad credential!")
|
|
# badCredentials = True
|
|
# time.sleep(2)
|
|
class CliInteraction(UserInteraction):
|
|
def prompt_up(self):
|
|
print("\nTouch your authenticator device now...\n")
|
|
|
|
def request_pin(self, permissions, rd_id):
|
|
return getpass("Enter PIN: ")
|
|
|
|
def request_uv(self, permissions, rd_id):
|
|
print("User Verification required.")
|
|
return True
|
|
|
|
|
|
|
|
while True:
|
|
# isNfc = False
|
|
if gpie:
|
|
GPIO.output(door, GPIO.LOW)
|
|
|
|
print("Waiting for device...")
|
|
badCredentials = False
|
|
if gpie:
|
|
GPIO.output(redLed, GPIO.HIGH)
|
|
while True:
|
|
try:
|
|
|
|
dev = next(CtapHidDevice.list_devices(), None)
|
|
|
|
if dev:
|
|
ctap2 = Ctap2(dev)
|
|
|
|
info = ctap2.get_info()
|
|
gotAaguid = info.aaguid
|
|
print(str(gotAaguid))
|
|
break
|
|
|
|
time.sleep(0.1)
|
|
except Exception as e:
|
|
print(e)
|
|
# Handle user interaction
|
|
|
|
try:
|
|
server = Fido2Server({"id": "example.com", "name": "Example RP"}, attestation="direct")
|
|
uv = "discouraged"
|
|
|
|
# Set up a FIDO 2 client using the origin https://example.com
|
|
client = Fido2Client(dev, "https://example.com", user_interaction=CliInteraction())
|
|
|
|
|
|
|
|
# Replace these variables with your connection parameters
|
|
|
|
user = dbusername
|
|
password = dbpassword
|
|
host = dbhost
|
|
|
|
# Define the SQL query for retrieving data
|
|
select_query = '''
|
|
SELECT "AAGUID", user_id, credential_id, pk_algo, pk_1, pk_3, pk_neg1, pk_neg2, pk_neg3, pickled
|
|
FROM credential_data
|
|
WHERE "AAGUID" = %s;
|
|
'''
|
|
|
|
|
|
# Function to connect to the PostgreSQL database and retrieve data
|
|
def fetch_data(aaguid):
|
|
conn = None
|
|
try:
|
|
# Connect to the PostgreSQL server
|
|
conn = psycopg2.connect(dbname=dbname, user=user, password=password, host=host)
|
|
cur = conn.cursor()
|
|
|
|
# Execute the SQL query
|
|
cur.execute(select_query, (gotAaguid,))
|
|
|
|
# Fetch the results
|
|
result = None
|
|
result = cur.fetchone()
|
|
if result:
|
|
data = {
|
|
"AAGUID": result[0],
|
|
"user_id": result[1],
|
|
"credential_id": result[2],
|
|
"pk_algo": result[3],
|
|
"pk_1": result[4],
|
|
"pk_3": result[5],
|
|
"pk_neg1": result[6],
|
|
"pk_neg2": result[7],
|
|
"pk_neg3": result[8],
|
|
"pickled": result[9]
|
|
}
|
|
return data
|
|
else:
|
|
print("No data found for AAGUID:", aaguid)
|
|
badCred()
|
|
raise Exception
|
|
return None
|
|
|
|
# Close communication with the database
|
|
cur.close()
|
|
except (Exception, psycopg2.DatabaseError) as error:
|
|
print(error)
|
|
raise Exception
|
|
finally:
|
|
if conn is not None:
|
|
conn.close()
|
|
|
|
|
|
# Example usage: Fetch data for a specific AAGUID and store in variables
|
|
aaguid_data = None
|
|
aaguid_data = fetch_data("")
|
|
user_id = aaguid_data["user_id"]
|
|
|
|
if aaguid_data:
|
|
|
|
# Store each piece of data into a separate variable
|
|
testa = aaguid_data["AAGUID"]
|
|
cred_id = aaguid_data["credential_id"]
|
|
pk_algo = aaguid_data["pk_algo"]
|
|
pk_1 = aaguid_data["pk_1"]
|
|
pk_3 = aaguid_data["pk_3"]
|
|
pk_neg1 = aaguid_data["pk_neg1"]
|
|
pk_neg2 = aaguid_data["pk_neg2"]
|
|
pk_neg3 = aaguid_data["pk_neg3"]
|
|
pickled = aaguid_data["pickled"]
|
|
|
|
if badCredentials:
|
|
raise Exception
|
|
|
|
credentials = pickle.loads(pickled)
|
|
|
|
|
|
# Prepare parameters for getAssertion
|
|
request_options, state = server.authenticate_begin(credentials, user_verification=uv)
|
|
|
|
# Authenticate the credential
|
|
|
|
result = client.get_assertion(request_options["publicKey"])
|
|
|
|
# Only one cred in allowCredentials, only one response.
|
|
result = result.get_response(0)
|
|
|
|
# Complete authenticator
|
|
server.authenticate_complete(
|
|
state,
|
|
|
|
credentials,
|
|
result.credential_id,
|
|
result.client_data,
|
|
result.authenticator_data,
|
|
result.signature,
|
|
)
|
|
|
|
print("Credential authenticated!")
|
|
print("CLIENT DATA:", result.client_data)
|
|
print()
|
|
print("AUTH DATA:", result.authenticator_data)
|
|
# GPIO.output(14, GPIO.LOW)
|
|
|
|
conn = None
|
|
try:
|
|
# Connect to the PostgreSQL server
|
|
conn = psycopg2.connect(dbname=dbname, user=user, password=password, host=host)
|
|
cur = conn.cursor()
|
|
select_query = '''
|
|
SELECT sec_level
|
|
FROM users
|
|
WHERE "user_id" = %s;
|
|
'''
|
|
# Execute the SQL query
|
|
cur.execute(select_query, (str(user_id)))
|
|
|
|
# Fetch the results
|
|
result = None
|
|
result = cur.fetchone()
|
|
if result:
|
|
userSecLevel = result[0]
|
|
|
|
else:
|
|
badCred()
|
|
raise Exception
|
|
|
|
# Close communication with the database
|
|
cur.close()
|
|
except (Exception, psycopg2.DatabaseError) as error:
|
|
print(error)
|
|
raise Exception
|
|
finally:
|
|
if conn is not None:
|
|
conn.close()
|
|
|
|
conn = None
|
|
try:
|
|
# Connect to the PostgreSQL server
|
|
conn = psycopg2.connect(dbname=dbname, user=user, password=password, host=host)
|
|
cur = conn.cursor()
|
|
select_query = '''
|
|
SELECT sec_level
|
|
FROM doors
|
|
WHERE "door_id" = %s;
|
|
'''
|
|
# Execute the SQL query
|
|
cur.execute(select_query, (doorID))
|
|
|
|
# Fetch the results
|
|
result = None
|
|
result = cur.fetchone()
|
|
if result:
|
|
doorSecLevel = result[0]
|
|
|
|
else:
|
|
badCred()
|
|
raise Exception
|
|
|
|
# Close communication with the database
|
|
cur.close()
|
|
except (Exception, psycopg2.DatabaseError) as error:
|
|
print(error)
|
|
raise Exception
|
|
finally:
|
|
if conn is not None:
|
|
conn.close()
|
|
|
|
if doorSecLevel < userSecLevel:
|
|
raise Exception
|
|
|
|
if gpie:
|
|
GPIO.output(redLed, GPIO.LOW)
|
|
for i in range(2):
|
|
GPIO.output(greenLed, GPIO.HIGH)
|
|
time.sleep(0.05)
|
|
GPIO.output(greenLed, GPIO.LOW)
|
|
time.sleep(0.05)
|
|
|
|
GPIO.output(greenLed, GPIO.HIGH)
|
|
GPIO.output(door, GPIO.HIGH)
|
|
time.sleep(3)
|
|
GPIO.output(greenLed, GPIO.LOW)
|
|
|
|
|
|
except Exception as e:
|
|
print("Authentication Failed!")
|
|
for i in range(5):
|
|
if gpie:
|
|
GPIO.output(redLed, GPIO.LOW)
|
|
time.sleep(0.05)
|
|
GPIO.output(redLed, GPIO.HIGH)
|
|
time.sleep(0.05)
|
|
|