malobeo/gatekeeper
3
0
Fork 0
Code Issues 7 Pull Requests 1 Actions Activity

just_found_out_about_linters (#17)

Browse Source 
This changed a bunch of code but no

Reviewed-on: #17
Co-authored-by: ahtlon <git@ahtlon.de>
Co-committed-by: ahtlon <git@ahtlon.de>
This commit was merged in pull request #17.
This commit is contained in:
ahtlon
2026-07-29 02:46:50 +02:00
committed by ahtlon
parent 683cd9a545
commit 3f20cdeed9
28 changed files with 736 additions and 564 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
app/services/auth.py
View File

@@ -1,18 +1,23 @@
import logging
logger = logging.getLogger(__name__)
import os
import secrets
import string
from datetime import UTC, datetime, timedelta
from typing import Annotated
from datetime import datetime, timedelta, timezone
from fastapi import APIRouter, HTTPException, Depends, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlmodel import Session, select
from pwdlib import PasswordHash
import jwt
from jwt.exceptions import InvalidTokenError
from app.model.models import UserDB, Token, TokenData, UserCreate
from app.services.database import *
import secrets, string, os
SECRET_KEY = os.getenv("SECRET_KEY", default="ff"*16)
import jwt
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from sqlmodel import Session, select
from app.model.models import Token, TokenData, UserDB
from app.services.database import add_and_refresh, get_session
logger = logging.getLogger(__name__)
SECRET_KEY = os.getenv("SECRET_KEY", default="ff" * 16)
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 120
@@ -22,16 +27,20 @@ token_router = APIRouter(tags=["Token"], prefix="/api/v1")
password_hash = PasswordHash.recommended()
def verify_password(plain_password, hashed_password):
return password_hash.verify(plain_password, hashed_password)
def get_password_hash(password):
return password_hash.hash(password)
def get_user(db, username: str):
user = db.exec(select(UserDB).where(UserDB.name == username)).first()
return user
def authenticate_user(db, username: str, password: str):
user = get_user(db, username)
if not user:
@@ -40,24 +49,26 @@ def authenticate_user(db, username: str, password: str):
return False
return user
def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.now(timezone.utc) + expires_delta
expire = datetime.now(UTC) + expires_delta
else:
expire = datetime.now(timezone.utc) + timedelta(minutes=15)
expire = datetime.now(UTC) + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def get_current_user(
token: Annotated[str, Depends(oauth2_scheme)],
db: Session = Depends(get_session),
):
):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"}
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
@@ -72,29 +83,29 @@ def get_current_user(
raise credentials_exception
return user
def auth_is_admin(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_session),
):
):
user = get_current_user(token=token, db=db)
if not user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to perform this action",
headers={"WWW-Authenticate": "Bearer"}
headers={"WWW-Authenticate": "Bearer"},
)
return True
def create_first_user(db: Session):
logger.info("Checking for admin user")
admin_user = db.exec(select(UserDB)).first()
if admin_user is None:
password = ''.join(secrets.choice(string.digits) for i in range(8))
password = "".join(secrets.choice(string.digits) for i in range(8))
logger.info(f"Creating first admin user with password: {password}")
user = UserDB(
name="admin",
passwordhash=get_password_hash(password),
is_admin=True
name="admin", passwordhash=get_password_hash(password), is_admin=True
)
return add_and_refresh(db, user)
logger.info(f"Admin user already exists: {admin_user.name}")
@@ -103,14 +114,14 @@ def create_first_user(db: Session):
@token_router.post("/token")
def login_for_access_token(
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
db: Session = Depends(get_session)
db: Session = Depends(get_session),
) -> Token:
user = authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or pw",
headers={"WWW-Authenticate": "Bearer"}
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
@@ -118,8 +129,7 @@ def login_for_access_token(
)
return Token(access_token=access_token, token_type="bearer")
@token_router.get("/test/login")
def test_login(
current_user: Annotated[UserDB, Depends(get_current_user)]
) -> UserDB:
return current_user
def test_login(current_user: Annotated[UserDB, Depends(get_current_user)]) -> UserDB:
return current_user

13
app/services/database.py
View File

@@ -1,19 +1,22 @@
import logging
logger = logging.getLogger(__name__)
from os import getenv, path
from sqlmodel import create_engine, SQLModel, Session
from app.model.models import Base
from sqlmodel import Session, SQLModel, create_engine
logger = logging.getLogger(__name__)
SQLALCHEMY_DATABASE_URL = getenv("SQLALCHEMY_DATABASE_URL", "sqlite:///./gatekeeper.db")
engine = create_engine(SQLALCHEMY_DATABASE_URL)
def create_db_and_tables():
if not path.exists(SQLALCHEMY_DATABASE_URL):
SQLModel.metadata.create_all(engine)
from alembic.config import Config
from alembic import command
alembic_cfg = Config("./alembic.ini")
command.stamp(alembic_cfg, "head")
logger.info("Database created and tables initialized.")
@@ -25,11 +28,13 @@ def get_session():
with Session(engine) as db:
yield db
def get_db_session():
return Session(engine)
def add_and_refresh(db: Session, obj):
db.add(obj)
db.commit()
db.refresh(obj)
return obj
return obj

36
app/services/door.py
View File

@@ -1,33 +1,35 @@
import logging
logger = logging.getLogger(__name__)
from sqlmodel import select
from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import selectinload
import sqlalchemy.exc as exc
from datetime import datetime, date, timedelta
from datetime import date, datetime, timedelta
from app.services.database import Session, get_session, add_and_refresh
from app.model.models import *
from sqlalchemy import exc
from sqlmodel import select
from app.model.models import Card, OneShotAccess
from app.services.database import Session, add_and_refresh
logger = logging.getLogger(__name__)
doorIsOpen = True
# I think this could also be gpio controlled
#See: https://github.com/technyon/nuki_hub#gpio-lock-control-optional
# See: https://github.com/technyon/nuki_hub#gpio-lock-control-optional
def openDoor():
global doorIsOpen
doorIsOpen = True
logger.info("Still needs gpio out")
pass
def closeDoor():
global doorIsOpen
doorIsOpen = False
logger.info("Still needs gpio out")
pass
def isDoorOpen():
return doorIsOpen
def decrementOneshot(db: Session, oneshot: OneShotAccess):
data = oneshot.model_dump()
if data["uses"] > 0:
@@ -35,6 +37,7 @@ def decrementOneshot(db: Session, oneshot: OneShotAccess):
oneshot.sqlmodel_update(oneshot, update=data)
add_and_refresh(db, oneshot)
def checkAccess(key: str, db: Session):
try:
current_weekday = datetime.weekday(date.today())
@@ -45,16 +48,20 @@ def checkAccess(key: str, db: Session):
if auth.type == "timetable":
for timetable in auth.timetables:
logger.info(f" checking timetable {timetable.id}")
logger.info(f" comparing weekday: CUR:{current_weekday} TT:{timetable.weekday}")
logger.info(
f" comparing weekday: CUR:{current_weekday} TT:{timetable.weekday}"
)
if current_weekday == timetable.weekday:
starttime = datetime.combine(date.today(), timetable.starttime)
endtime = starttime + timedelta(minutes=timetable.duration)
logger.info(f" comparing time: Start:{starttime} Current:{current_time} End:{endtime}")
logger.info(
f" comparing time: Start:{starttime} Current:{current_time} End:{endtime}"
)
if starttime < current_time < endtime:
logger.info("Access Valid!")
return True
if auth.type == "oneshot":
logger.info(f' oneshot auth found: {auth.oneshot}')
logger.info(f" oneshot auth found: {auth.oneshot}")
if current_time < auth.oneshot.ends_at:
if auth.oneshot.uses > 0:
decrementOneshot(db, auth.oneshot)
@@ -63,4 +70,3 @@ def checkAccess(key: str, db: Session):
return False
except exc.NoResultFound:
raise Exception("No Access with that key found, this might be a db error")

144
app/services/scanner.py
View File

@@ -1,33 +1,39 @@
import logging
logger = logging.getLogger(__name__)
import threading
import time
import os
import secrets
import threading
import time
from typing import Optional
from sqlmodel import Session
from desfire import (
DESFire,
DESFireKey,
PCSCDevice,
diversify_key,
get_list,
to_hex_string,
)
from desfire.enums import (
DESFireCommunicationMode,
DESFireFileType,
DESFireKeySettings,
DESFireKeyType,
)
from desfire.schemas import FilePermissions, FileSettings, KeySettings
from dotenv import load_dotenv
from fastapi import HTTPException, status
from smartcard.CardRequest import CardRequest
from smartcard.CardType import AnyCardType
from smartcard.Exceptions import CardRequestTimeoutException
from desfire import DESFire, DESFireKey, PCSCDevice, diversify_key, get_list, to_hex_string
from desfire.enums import DESFireCommunicationMode, DESFireFileType, DESFireKeySettings, DESFireKeyType
from desfire.schemas import FilePermissions, FileSettings, KeySettings
import desfire.exceptions as desExceptions
from app.services.door import checkAccess, openDoor
from app.services.door import openDoor, closeDoor, isDoorOpen, checkAccess
logger = logging.getLogger(__name__)
#ENV vars
# ENV vars
load_dotenv()
MIFARE_APP_MASTER_KEY = os.getenv('MIFARE_APP_MASTER_KEY')
MIFARE_ACL_READ_BASE_KEY = os.getenv('MIFARE_ACL_READ_BASE_KEY')
MIFARE_ACL_WRITE_BASE_KEY = os.getenv('MIFARE_ACL_WRITE_BASE_KEY')
MIFARE_APP_MASTER_KEY = os.getenv("MIFARE_APP_MASTER_KEY")
MIFARE_ACL_READ_BASE_KEY = os.getenv("MIFARE_ACL_READ_BASE_KEY")
MIFARE_ACL_WRITE_BASE_KEY = os.getenv("MIFARE_ACL_WRITE_BASE_KEY")
# Constants
MIFARE_APP_ID = "DEAFFE" # 7 bytes
@@ -36,10 +42,15 @@ MIFARE_ACL_WRITE_BASE_KEY_ID = 0x2
MIFARE_SYS_ID = "FF0000" # 3 bytes, can essentially be anything
MIFARE_ENCRYPTED_FILE_ID = 0x1
def checkForKey():
if MIFARE_APP_MASTER_KEY == None:
logger.critical("NO MASTER KEY LOADED")
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="No key loaded! Check application.")
if MIFARE_APP_MASTER_KEY is None:
logger.critical("NO MASTER KEY LOADED")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="No key loaded! Check application.",
)
def getCardService(timeout: int = 10):
cardtype = AnyCardType()
@@ -48,51 +59,58 @@ def getCardService(timeout: int = 10):
cardservice.connection.connect()
return cardservice
def readFileOnCard(desfire: DESFire):
if not MIFARE_ACL_READ_BASE_KEY:
logger.critical("MIFARE_ACL_READ_BASE_KEY not found! Reading skipped!")
return
#create keys
#desfire = DESFire(PCSCDevice(cardservice.connection.component))
# create keys
# desfire = DESFire(PCSCDevice(cardservice.connection.component))
aes_keysettings = KeySettings(key_type=DESFireKeyType.DF_KEY_AES)
keysettings = desfire.get_key_setting()
desKey = DESFireKey(keysettings, "00" * 8)
# Get real UID
desfire.authenticate(0x0, desKey)
#To get the uid you have to auth with an empty (default) key
# To get the uid you have to auth with an empty (default) key
uid = desfire.get_real_uid()
applications = desfire.get_application_ids()
try:
assert len(applications) == 1
assert applications[0] == get_list(MIFARE_APP_ID)
assert applications[0] == get_list(MIFARE_APP_ID)
except AssertionError:
logger.error("No application found!")
time.sleep(4)
return
#Then use the key derivation with that uid, the appid, the sysid
diversification_data = [0x01] + uid + get_list(MIFARE_APP_ID) + get_list(MIFARE_SYS_ID)
read_div_key_bytes = diversify_key(get_list(MIFARE_ACL_READ_BASE_KEY), diversification_data, pad_to_32=False)
#Log in with derived read key
# Then use the key derivation with that uid, the appid, the sysid
diversification_data = (
[0x01] + uid + get_list(MIFARE_APP_ID) + get_list(MIFARE_SYS_ID)
)
read_div_key_bytes = diversify_key(
get_list(MIFARE_ACL_READ_BASE_KEY), diversification_data, pad_to_32=False
)
# Log in with derived read key
logger.debug("Start auth")
aes_app_read_key = DESFireKey(aes_keysettings, read_div_key_bytes)
desfire.select_application(MIFARE_APP_ID)
desfire.authenticate(MIFARE_ACL_READ_BASE_KEY_ID, aes_app_read_key)
logger.debug(f"Read data from {MIFARE_ENCRYPTED_FILE_ID}")
file_data = desfire.get_file_settings(MIFARE_ENCRYPTED_FILE_ID)
rdata = desfire.read_file_data(MIFARE_ENCRYPTED_FILE_ID, file_data)
#convert list of int to str
# convert list of int to str
rdata = to_hex_string(rdata).replace(" ", "").lower()
logger.debug(f"Data on card: {rdata}")
return rdata
def DeleteCard():
try:
checkForKey()
from app.main import scanner as scannerThread
scannerThread.stop()
cardservice = getCardService(15)
@@ -107,12 +125,12 @@ def DeleteCard():
desKey = DESFireKey(des_keysettings, "00" * 8)
aes_master_key = DESFireKey(aes_keysettings, MIFARE_APP_MASTER_KEY)
aes_null_key = DESFireKey(aes_keysettings, "00" * 16)
desfire.select_application(0x0)
try:
try:
logger.debug("Auth1")#
logger.debug("Auth1")
desfire.authenticate(0x0, aes_master_key)
except:
logger.debug("Auth2")
@@ -120,11 +138,13 @@ def DeleteCard():
except:
logger.debug("Auth3")
desfire.authenticate(0x0, desKey)
applications = desfire.get_application_ids()
logger.debug(f"Applications: {applications}")
if len(applications) == 0:
raise HTTPException(status_code=status.HTTP_410_GONE, detail="No applications on card")
raise HTTPException(
status_code=status.HTTP_410_GONE, detail="No applications on card"
)
desfire.select_application(MIFARE_APP_ID)
desfire.authenticate(0x0, aes_master_key)
@@ -136,14 +156,18 @@ def DeleteCard():
pass
scannerThread.start()
return rdata
except(Exception, AssertionError) as e:
except (Exception, AssertionError) as e:
logger.error(f"Error in deletion function: {e}", exc_info=True)
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error: {e}"
)
def WriteNewCard():
try:
checkForKey()
from app.main import scanner as scannerThread
scannerThread.stop()
cardservice = getCardService(20)
@@ -159,8 +183,8 @@ def WriteNewCard():
# Authenticate with default DES key
logger.debug("Authenticating with default DES key...")
desfire.authenticate(0x0, desKey)
#get uid
# get uid
uid = desfire.get_real_uid()
# Set default key
@@ -189,7 +213,7 @@ def WriteNewCard():
# Select application
desfire.select_application(MIFARE_APP_ID)
#recreate key object
# recreate key object
desfire.authenticate(0x0, aes_null_key)
desfire.change_key(0x0, aes_null_key, aes_master_key, 0x1)
@@ -198,18 +222,28 @@ def WriteNewCard():
aes_null_key = DESFireKey(aes_keysettings, "00" * 16)
#generate div data
diversification_data = [0x01] + uid + get_list(MIFARE_APP_ID) + get_list(MIFARE_SYS_ID)
read_div_key_bytes = diversify_key(get_list(MIFARE_ACL_READ_BASE_KEY), diversification_data, pad_to_32=False)
write_div_key_bytes = diversify_key(get_list(MIFARE_ACL_WRITE_BASE_KEY), diversification_data, pad_to_32=False)
# generate div data
diversification_data = (
[0x01] + uid + get_list(MIFARE_APP_ID) + get_list(MIFARE_SYS_ID)
)
read_div_key_bytes = diversify_key(
get_list(MIFARE_ACL_READ_BASE_KEY), diversification_data, pad_to_32=False
)
write_div_key_bytes = diversify_key(
get_list(MIFARE_ACL_WRITE_BASE_KEY), diversification_data, pad_to_32=False
)
logger.debug("Changing file read key...")
aes_file_read_key = DESFireKey(aes_keysettings, read_div_key_bytes)
desfire.change_key(MIFARE_ACL_READ_BASE_KEY_ID, aes_null_key, aes_file_read_key, 0x1)
desfire.change_key(
MIFARE_ACL_READ_BASE_KEY_ID, aes_null_key, aes_file_read_key, 0x1
)
logger.debug("Changing file write key...")
aes_file_write_key = DESFireKey(aes_keysettings, write_div_key_bytes)
desfire.change_key(MIFARE_ACL_WRITE_BASE_KEY_ID, aes_null_key, aes_file_write_key, 0x1)
desfire.change_key(
MIFARE_ACL_WRITE_BASE_KEY_ID, aes_null_key, aes_file_write_key, 0x1
)
logger.debug("Create encrypted file containing key...")
file_settings = FileSettings(
@@ -226,25 +260,29 @@ def WriteNewCard():
logger.debug("Writing UID to encrypted file...")
key = secrets.token_hex(16)
desfire.write_file_data(MIFARE_ENCRYPTED_FILE_ID, 0x0, file_data.encryption, get_list(key))
desfire.write_file_data(
MIFARE_ENCRYPTED_FILE_ID, 0x0, file_data.encryption, get_list(key)
)
logger.debug("Reading from encrypted file...")
rdata = desfire.read_file_data(MIFARE_ENCRYPTED_FILE_ID, file_data)
assert rdata == get_list(key)
logger.debug(" - Data written successfully.")
scannerThread.start()
return key, to_hex_string(data=uid, separator=":")
return key, to_hex_string(data=uid, separator=":")
except Exception as e:
logger.error(f"Error in write function: {e}", exc_info=True)
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error: {e}"
)
class BackgroundScanner:
def __init__(self, db):
self.db = db
self.is_running = False
self.thread: Optional[threading.Thread] = None
self.thread: threading.Thread | None = None
def start(self):
if self.is_running:
@@ -292,10 +330,10 @@ class BackgroundScanner:
except Exception as e:
logger.error(f"something went wrong: {e}")
time.sleep(5)
def _check_db(self, key):
check = checkAccess(key, self.db)
if check == True:
if check:
openDoor()
logger.info("Access granted!")
else:

29
app/services/settings.py
View File

@@ -1,26 +1,29 @@
import os
import logging
import os
logger = logging.getLogger(__name__)
def verify_settings():
card_envs = [
card_envs = [
"MIFARE_APP_MASTER_KEY",
"MIFARE_ACL_READ_BASE_KEY",
"MIFARE_ACL_WRITE_BASE_KEY",
]
important_envs = [
"SECRET_KEY"
]
other_envs = [
"SQLALCHEMY_DATABASE_URL"
]
important_envs = ["SECRET_KEY"]
other_envs = ["SQLALCHEMY_DATABASE_URL"]
for setting in card_envs:
if (setting not in os.environ or setting == "") and not os.getenv("DISABLE_CARDS"):
raise ValueError(f"Missing environment variable for scanner start: {setting} \n Run with DISABLE_CARDS env var to disable cards")
if (setting not in os.environ or setting == "") and not os.getenv(
"DISABLE_CARDS"
):
raise ValueError(
f"Missing environment variable for scanner start: {setting} \n Run with DISABLE_CARDS env var to disable cards"
)
for setting in important_envs:
if setting not in os.environ or setting == "":
raise ValueError(f'Missing critical environment variable {setting}. Stopping...')
raise ValueError(
f"Missing critical environment variable {setting}. Stopping..."
)
for setting in other_envs:
if setting not in os.environ:
logger.critical(f'Env var {setting} not set. Continuing with defaults.')
logger.critical(f"Env var {setting} not set. Continuing with defaults.")