From ede3e9a516e2e64d2d02cd751c9b847e3bcb03e1 Mon Sep 17 00:00:00 2001 From: ahtlon Date: Wed, 29 Jul 2026 01:31:41 +0200 Subject: [PATCH 1/9] [uv] add ruff (the linter) --- pyproject.toml | 1 + uv.lock | 27 +++++++++++++++++++++++++++ 2 files changed, 28 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index 9fe9cf6..9244d55 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,7 @@ dependencies = [ "setuptools>=82.0.1", "pyscard>=2.3.1", "alembic>=1.18.5", + "ruff>=0.16.0", ] [tool.uv.sources] diff --git a/uv.lock b/uv.lock index da3ea75..caa56d5 100644 --- a/uv.lock +++ b/uv.lock @@ -632,6 +632,7 @@ dependencies = [ { name = "pytest-cov" }, { name = "python-desfire" }, { name = "requests" }, + { name = "ruff" }, { name = "setuptools" }, { name = "sqlmodel" }, ] @@ -648,6 +649,7 @@ requires-dist = [ { name = "pytest-cov", specifier = ">=7.1.0" }, { name = "python-desfire", git = "https://github.com/waza-ari/python-desfire" }, { name = "requests", specifier = ">=2.33.1" }, + { name = "ruff", specifier = ">=0.16.0" }, { name = "setuptools", specifier = ">=82.0.1" }, { name = "sqlmodel", specifier = ">=0.0.38" }, ] @@ -1537,6 +1539,31 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/79/62/b88e5879512c55b8ee979c666ee6902adc4ed05007226de266410ae27965/rignore-0.7.6-cp314-cp314t-win_arm64.whl", hash = "sha256:b83adabeb3e8cf662cabe1931b83e165b88c526fa6af6b3aa90429686e474896", size = 656035, upload-time = "2025-11-05T21:41:31.13Z" }, ] +[[package]] +name = "ruff" +version = "0.16.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/4d/94/1e5e4967626faf12fa56999cd6222dff6992ceb086ad7945756baf70c7a7/ruff-0.16.0.tar.gz", hash = "sha256:e460aafd5495ec89efaa6ced2e4a9a581116451e1c88b9d37ef497e0f8e93982", size = 4790557, upload-time = "2026-07-23T19:11:30.981Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/4b/81/1c8818fee7ce1a04cd7d1b3172e0a8f8e4f1dc4feb7fc390e16daa8af323/ruff-0.16.0-py3-none-linux_armv6l.whl", hash = "sha256:e5115729eb08c585e5121978ba5d5b60caeae394ce21b9fb5e6cd33a1c6c9b1e", size = 10754633, upload-time = "2026-07-23T19:10:46.415Z" }, + { url = "https://files.pythonhosted.org/packages/23/df/beaf59c09d68db84304d555f188b276a77132a5d5b0b67a5c762aa143628/ruff-0.16.0-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:3c954b1d580bfa035b41654f7858cc7e71d5fc3ac5b723dd62bd9133830ed522", size = 10969164, upload-time = "2026-07-23T19:10:50.271Z" }, + { url = "https://files.pythonhosted.org/packages/42/ce/741cd197496a1abbf51352710fd15ed995d2a2be87189c1da26a450d6e83/ruff-0.16.0-py3-none-macosx_11_0_arm64.whl", hash = "sha256:e01c21d10eb1b29f47b7454e1f4056db9a3f0260c646aa88457c610291db9f81", size = 10488846, upload-time = "2026-07-23T19:10:52.639Z" }, + { url = "https://files.pythonhosted.org/packages/52/2a/a2db8e88cade358f5cdcb05674a917751074109315d014eb6352d9a893f7/ruff-0.16.0-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:6e364e5ed22ed8dc05082fd78e35308618260907ac2d3c1d637b2e682415b6c9", size = 10889729, upload-time = "2026-07-23T19:10:54.89Z" }, + { url = "https://files.pythonhosted.org/packages/42/65/62a771694ebd63029dc953e27dbad40e1588bd4860ff9fe881018fddaa49/ruff-0.16.0-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:d327b8fc113a1d4421a04f3839d3752057c8dd1ee320223a6f3f52d04ada462a", size = 10568275, upload-time = "2026-07-23T19:10:56.993Z" }, + { url = "https://files.pythonhosted.org/packages/3f/e2/ced249fe8af5f086c5c58cc21cc3356d50f32f7401c5df87050c999620a7/ruff-0.16.0-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:a9b50c55e263103586b3dcf5f73d479eb8cb5fdb6098fec59a62891dab653717", size = 11385112, upload-time = "2026-07-23T19:10:59.615Z" }, + { url = "https://files.pythonhosted.org/packages/87/0b/05154977a8fd69eeb6c103271f55403bfd8711f5c0f8ed07489d95a504e7/ruff-0.16.0-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:0ff4a79ce3ec0172f3241943835de1c4cb4e2dcd07f0f8c2d02603dbbbee4b17", size = 12207008, upload-time = "2026-07-23T19:11:02.154Z" }, + { url = "https://files.pythonhosted.org/packages/fb/29/98225831a3a1eab0e02f4acc6ca6559a98611dcc68b6965ff4b7234627c1/ruff-0.16.0-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:e95c448fca1fb2a18372a9440926c5a6ee789639bb975c72e7ae6d0b04218ab4", size = 11650842, upload-time = "2026-07-23T19:11:04.557Z" }, + { url = "https://files.pythonhosted.org/packages/91/66/6bd3cf90500653d55dc0ffc8507aa8300bd49d0214b2e8cb4d3fef2943ba/ruff-0.16.0-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:4f11a8d11010301d0a398a2fdef67691feca7294da6aef55e2150e8fa2cd520b", size = 11400718, upload-time = "2026-07-23T19:11:09.233Z" }, + { url = "https://files.pythonhosted.org/packages/8e/a2/a54eb4eae05d66364050a5d3b8a9c5ef88196531b3cbe7109d873f87f819/ruff-0.16.0-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:48044c678e9cb8698246c99b14aaccfa6601dea7379eb48a6f8f73f7a6d86cd0", size = 11426177, upload-time = "2026-07-23T19:11:11.994Z" }, + { url = "https://files.pythonhosted.org/packages/1a/be/16e3eea4b2a478a496919f5e36f17c4559e54620bd3bbac5d6affa068006/ruff-0.16.0-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:7aa0959bad8eb8bef50340154fc9b58678dae31fa4293afa38b44b6e552c0213", size = 10856126, upload-time = "2026-07-23T19:11:14.221Z" }, + { url = "https://files.pythonhosted.org/packages/a2/84/252eb8b868a16eec7257c14f504f77537e734b2d69c762e639e588e304a3/ruff-0.16.0-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:28ea2b7df8ebf7f9da6b7d47b230ab48f387c0a29be3b474c4d0740e197bb9af", size = 10571208, upload-time = "2026-07-23T19:11:16.378Z" }, + { url = "https://files.pythonhosted.org/packages/21/09/817a482f542f7570cbb4554b26e896610c7114f539b1d9e2d2145bf6bef6/ruff-0.16.0-py3-none-musllinux_1_2_i686.whl", hash = "sha256:33a3dfac8c35f81498dea9181bccc2f4c4bc8f1521a1dd9406e77643e0f0fb09", size = 11063329, upload-time = "2026-07-23T19:11:19.173Z" }, + { url = "https://files.pythonhosted.org/packages/2e/23/9403c180ca1cb9b1f7335f5c3e5305c09d49ea5b345196682a36028bde4a/ruff-0.16.0-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:a5237a0bda500d30d81b8e07a6973a5cbc772864cbf746ae2f4e8a2e01c9f4ed", size = 11489751, upload-time = "2026-07-23T19:11:21.74Z" }, + { url = "https://files.pythonhosted.org/packages/b2/1d/1b2ef7bcde851c78d7f17f1cca13fd6dc695fc4b3d6197941e72cae5b132/ruff-0.16.0-py3-none-win32.whl", hash = "sha256:7fab76fa065c873f41ff744347c6e77bcc3dfec4bcc754dc26b63d23c0f7f5fb", size = 10785885, upload-time = "2026-07-23T19:11:23.947Z" }, + { url = "https://files.pythonhosted.org/packages/b2/a3/d5e4ef7a56be3f928ffb90b94c25ba7d3cb9c7fe0736aeaaedf361770712/ruff-0.16.0-py3-none-win_amd64.whl", hash = "sha256:429c117f022bf481fabd9d551e7a3952b24c65e6ef44337ea09d90bebef14472", size = 11923141, upload-time = "2026-07-23T19:11:26.409Z" }, + { url = "https://files.pythonhosted.org/packages/cb/9a/8415f2657cbe200f41a4531ccededf135505a92d4a012229121f885b26f9/ruff-0.16.0-py3-none-win_arm64.whl", hash = "sha256:14296fedcd2705c77ab8235439278bbb38f285cf7da5528b00b3e330c3d4872d", size = 11273407, upload-time = "2026-07-23T19:11:28.705Z" }, +] + [[package]] name = "secretstorage" version = "3.5.0" -- 2.51.2 From 1ca046ded9ec9ec8a538068b10e17d9208f9cc7f Mon Sep 17 00:00:00 2001 From: ahtlon Date: Wed, 29 Jul 2026 01:41:26 +0200 Subject: [PATCH 2/9] wow, thats pretty ruff ran 'ruff check --fix app' --- app/__init__.py | 3 ++- app/controllers/aaManager.py | 12 +++++------ app/controllers/cardManager.py | 22 ++++++++++----------- app/controllers/debugManager.py | 10 ++++++---- app/controllers/doorManager.py | 6 +++--- app/controllers/groupManager.py | 12 +++++------ app/controllers/userManager.py | 18 +++++++++-------- app/main.py | 21 ++++++++++++++------ app/model/models.py | 26 +++++++++++++----------- app/services/auth.py | 23 +++++++++++++--------- app/services/database.py | 5 +++-- app/services/door.py | 16 +++++++-------- app/services/scanner.py | 35 ++++++++++++++++++++------------- app/services/settings.py | 3 ++- 14 files changed, 120 insertions(+), 92 deletions(-) diff --git a/app/__init__.py b/app/__init__.py index 958d6b5..1217788 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -1,5 +1,6 @@ from fastapi import FastAPI -from .controllers import userManager, cardManager + +from .controllers import cardManager, userManager app = FastAPI() app.include_router(userManager.user_router) diff --git a/app/controllers/aaManager.py b/app/controllers/aaManager.py index 2fa2e8c..ad31e64 100644 --- a/app/controllers/aaManager.py +++ b/app/controllers/aaManager.py @@ -1,21 +1,21 @@ import logging + logger = logging.getLogger(__name__) + from fastapi import APIRouter, Depends, HTTPException, status -from sqlmodel import Session, select from sqlalchemy.orm import selectinload -from typing import List +from sqlmodel import Session, select from app.model.models import * -from app.services.database import engine, get_session, add_and_refresh from app.services.auth import auth_is_admin -import uuid as gen_uuid +from app.services.database import add_and_refresh, get_session aa_router = APIRouter(prefix="/api/v1/aa", tags=["AccessAuth"]) @aa_router.post("/", response_model=AccessAuthorizationResponse) def add_accessauth(*, db: Session = Depends(get_session), aa: AccessAuthorizationCreate, admin: bool = Depends(auth_is_admin)): logger.info(f"Creating accessauth with data: {aa}") - if aa.timetables is not []: + if aa.timetables != []: timetables = [Timetable.model_validate(t) for t in aa.timetables] else: timetables = [] if aa.oneshot is not None: @@ -30,7 +30,7 @@ def add_accessauth(*, db: Session = Depends(get_session), aa: AccessAuthorizatio ) return add_and_refresh(db, db_aa) -@aa_router.get("/", response_model=List[AccessAuthorizationResponse]) +@aa_router.get("/", response_model=list[AccessAuthorizationResponse]) def get_all_accessauths(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): return db.exec( select(AccessAuthorizationDB) diff --git a/app/controllers/cardManager.py b/app/controllers/cardManager.py index 0e9f527..68cba9d 100644 --- a/app/controllers/cardManager.py +++ b/app/controllers/cardManager.py @@ -1,15 +1,15 @@ import logging -logger = logging.getLogger(__name__) -from fastapi import APIRouter, Depends, HTTPException, status -from sqlmodel import Session, select -from typing import List -from sqlalchemy.exc import NoResultFound -from ..model.models import Card, CardCreate, CardUpdate, GroupDB -from ..services.database import engine, get_session, add_and_refresh -from ..services.auth import auth_is_admin -import uuid as gen_uuid -from app.services.scanner import WriteNewCard, DeleteCard +logger = logging.getLogger(__name__) + +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy.exc import NoResultFound +from sqlmodel import Session, select + +from app.model.models import Card, CardCreate, CardUpdate, GroupDB +from app.services.auth import auth_is_admin +from app.services.database import add_and_refresh, get_session +from app.services.scanner import DeleteCard, WriteNewCard card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"]) @@ -51,7 +51,7 @@ def del_card(*, db: Session = Depends(get_session), admin: bool = Depends(auth_i db.commit() return {"message": "Card deleted successfully"} -@card_router.get("/{group_id}", response_model=List[Card]) +@card_router.get("/{group_id}", response_model=list[Card]) def get_cards(*, db: Session = Depends(get_session), group_id: int, admin: bool = Depends(auth_is_admin)): cards = db.exec(select(Card).where(Card.group_id == group_id)).all() return cards diff --git a/app/controllers/debugManager.py b/app/controllers/debugManager.py index e8502e4..0f6addf 100644 --- a/app/controllers/debugManager.py +++ b/app/controllers/debugManager.py @@ -1,11 +1,13 @@ import logging + logger = logging.getLogger(__name__) from fastapi import APIRouter, Depends, HTTPException -from app.services.auth import auth_is_admin +from sqlalchemy import exc from sqlmodel import Session, select -import sqlalchemy.exc as exc + from app.model.models import * -from app.services.database import get_session, add_and_refresh +from app.services.auth import auth_is_admin +from app.services.database import add_and_refresh, get_session debug_router = APIRouter( prefix="/api/v1/debug", @@ -41,7 +43,7 @@ def remove_card_manually(card_id: str, db: Session = Depends(get_session)): @debug_router.put("/getcards") def list_all_cards(db: Session = Depends(get_session)): - logger.info(f"Debug Setting: Getting cards.") + logger.info("Debug Setting: Getting cards.") cards = db.exec(select(Card)).all() print(cards) out = [] diff --git a/app/controllers/doorManager.py b/app/controllers/doorManager.py index 3e419b9..bfe22d0 100644 --- a/app/controllers/doorManager.py +++ b/app/controllers/doorManager.py @@ -1,9 +1,9 @@ -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends from sqlmodel import Session -from app.services.database import get_session -from app.services.auth import auth_is_admin import app.services.door as doorService +from app.services.auth import auth_is_admin +from app.services.database import get_session door_router = APIRouter(prefix="/api/v1/door",tags=["Door"]) diff --git a/app/controllers/groupManager.py b/app/controllers/groupManager.py index d5a4800..88a6786 100644 --- a/app/controllers/groupManager.py +++ b/app/controllers/groupManager.py @@ -1,14 +1,14 @@ -from fastapi import APIRouter, HTTPException, Depends, status -from sqlmodel import Session, select -from typing import List -from ..model.models import GroupDB, GroupResponse, GroupCreate -from ..services.database import engine, get_session, add_and_refresh +from fastapi import APIRouter, Depends, HTTPException, status +from sqlmodel import Session, select + +from ..model.models import GroupCreate, GroupDB, GroupResponse from ..services.auth import auth_is_admin +from ..services.database import add_and_refresh, get_session group_router = APIRouter(prefix="/api/v1/groups", tags=["Group"]) -@group_router.get("/", response_model=List[GroupResponse]) +@group_router.get("/", response_model=list[GroupResponse]) def get_groups(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): groups = db.exec(select(GroupDB)).all() return groups diff --git a/app/controllers/userManager.py b/app/controllers/userManager.py index 22bfbb4..f19a753 100644 --- a/app/controllers/userManager.py +++ b/app/controllers/userManager.py @@ -1,12 +1,14 @@ import logging -logger = logging.getLogger(__name__) -from fastapi import APIRouter, HTTPException, Depends, status -from sqlmodel import Session, select -from typing import List -from ..model.models import UserResponse, UserCreate, UserDB, UserUpdate -from ..services.database import engine, get_session, add_and_refresh -from ..services.auth import get_password_hash, get_current_user as auth_user, auth_is_admin +logger = logging.getLogger(__name__) + +from fastapi import APIRouter, Depends, HTTPException, status +from sqlmodel import Session, select + +from ..model.models import UserCreate, UserDB, UserResponse, UserUpdate +from ..services.auth import auth_is_admin, get_password_hash +from ..services.auth import get_current_user as auth_user +from ..services.database import add_and_refresh, get_session user_router = APIRouter(tags=["Users"], prefix="/api/v1/users") @@ -19,7 +21,7 @@ def create_user(*, db: Session = Depends(get_session), user: UserCreate, admin: db_user = UserDB.model_validate(user, update=hashed_password) return add_and_refresh(db, db_user) -@user_router.get("/", response_model=List[UserResponse]) +@user_router.get("/", response_model=list[UserResponse]) def read_users(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): users = db.exec(select(UserDB)).all() return users diff --git a/app/main.py b/app/main.py index 691e532..3e6dde4 100644 --- a/app/main.py +++ b/app/main.py @@ -1,19 +1,28 @@ import logging + logger = logging.getLogger(__name__) import os -from fastapi import FastAPI -from fastapi.middleware.cors import CORSMiddleware -from fastapi.security import OAuth2PasswordBearer from contextlib import asynccontextmanager from dotenv import load_dotenv +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware +from fastapi.security import OAuth2PasswordBearer + load_dotenv() -from app.controllers import userManager, cardManager, groupManager, aaManager, doorManager, debugManager +from app.controllers import ( + aaManager, + cardManager, + debugManager, + doorManager, + groupManager, + userManager, +) +from app.services.auth import create_first_user, token_router from app.services.database import create_db_and_tables, get_db_session -from app.services.auth import token_router, create_first_user -from app.services.settings import verify_settings from app.services.scanner import BackgroundScanner +from app.services.settings import verify_settings oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") scanner = BackgroundScanner(db=get_db_session()) diff --git a/app/model/models.py b/app/model/models.py index b198e15..253887a 100644 --- a/app/model/models.py +++ b/app/model/models.py @@ -1,7 +1,9 @@ -from sqlmodel import Field, Relationship, Session, SQLModel -from typing import List, Literal, Union from datetime import datetime, time +from typing import Literal + from pydantic import model_validator +from sqlmodel import Field, Relationship, SQLModel + class Base(SQLModel): pass @@ -50,13 +52,13 @@ class GroupCreate(GroupBase): class GroupDB(GroupBase, table=True): id: int | None = Field(default=None, primary_key=True) - cards: List["Card"] = Relationship(back_populates="group") - accessauths: List["AccessAuthorizationDB"] = Relationship(back_populates="groups", link_model=AaGroupLink) + cards: list["Card"] = Relationship(back_populates="group") + accessauths: list["AccessAuthorizationDB"] = Relationship(back_populates="groups", link_model=AaGroupLink) class GroupResponse(GroupBase): id: int - cards: List["Card"] | None - accessauths: List["AccessAuthorizationDB"] | None + cards: list["Card"] | None + accessauths: list["AccessAuthorizationDB"] | None #### AccessAuthorization class AccessAuthorizationBase(Base): @@ -67,8 +69,8 @@ class AccessAuthorizationBase(Base): class AccessAuthorizationDB(AccessAuthorizationBase, table=True): id: int | None = Field(default=None, primary_key=True) type: str - groups: List["GroupDB"] = Relationship(back_populates="accessauths", link_model=AaGroupLink) - timetables: List["Timetable"] = Relationship(back_populates="accessauth", cascade_delete=True) + groups: list["GroupDB"] = Relationship(back_populates="accessauths", link_model=AaGroupLink) + timetables: list["Timetable"] = Relationship(back_populates="accessauth", cascade_delete=True) oneshot: "OneShotAccess" = Relationship(back_populates="accessauth", cascade_delete=True) class OneShotAccessBase(Base): @@ -81,7 +83,7 @@ class OneShotAccess(OneShotAccessBase, table=True): accessauth: AccessAuthorizationDB = Relationship(back_populates="oneshot") class AccessAuthorizationCreate(AccessAuthorizationBase): - timetables: List["TimetableCreate"] = [] + timetables: list["TimetableCreate"] = [] oneshot: OneShotAccessBase | None = None @model_validator(mode="after") @@ -102,15 +104,15 @@ class AccessAuthorizationCreate(AccessAuthorizationBase): class AccessAuthorizationResponse(AccessAuthorizationBase): id: int - timetables: List["Timetable"] = [] + timetables: list["Timetable"] = [] oneshot: OneShotAccessBase | None = None - groups: List["GroupDB"] + groups: list["GroupDB"] class AccessAuthorizationUpdate(Base): name: str | None = None type: Literal["timetable", "oneshot", "somefuturespec"] | None = None is_active: bool | None = None - timetables: List["TimetableCreate"] | None = None + timetables: list["TimetableCreate"] | None = None oneshot: OneShotAccessBase | None = None diff --git a/app/services/auth.py b/app/services/auth.py index 55e9590..619c0d0 100644 --- a/app/services/auth.py +++ b/app/services/auth.py @@ -1,16 +1,21 @@ 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 fastapi import APIRouter, Depends, HTTPException, status +from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from jwt.exceptions import InvalidTokenError -from app.model.models import UserDB, Token, TokenData, UserCreate +from pwdlib import PasswordHash +from sqlmodel import Session, select + +from app.model.models import Token, TokenData, UserDB from app.services.database import * -import secrets, string, os SECRET_KEY = os.getenv("SECRET_KEY", default="ff"*16) ALGORITHM = "HS256" @@ -43,9 +48,9 @@ def authenticate_user(db, username: str, password: str): 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 diff --git a/app/services/database.py b/app/services/database.py index 1f065cd..447fca3 100644 --- a/app/services/database.py +++ b/app/services/database.py @@ -1,9 +1,9 @@ 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 SQLALCHEMY_DATABASE_URL = getenv("SQLALCHEMY_DATABASE_URL", "sqlite:///./gatekeeper.db") @@ -13,6 +13,7 @@ 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") diff --git a/app/services/door.py b/app/services/door.py index f4d3548..93c508b 100644 --- a/app/services/door.py +++ b/app/services/door.py @@ -1,13 +1,13 @@ 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 app.services.database import Session, get_session, add_and_refresh +logger = logging.getLogger(__name__) +from datetime import date, datetime, timedelta + +from sqlalchemy import exc +from sqlmodel import select + from app.model.models import * +from app.services.database import Session, add_and_refresh doorIsOpen = True # I think this could also be gpio controlled @@ -17,13 +17,11 @@ 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 diff --git a/app/services/scanner.py b/app/services/scanner.py index 56bbf49..75e1fa1 100644 --- a/app/services/scanner.py +++ b/app/services/scanner.py @@ -1,27 +1,34 @@ 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 openDoor, closeDoor, isDoorOpen, checkAccess - +from app.services.door import checkAccess, openDoor #ENV vars load_dotenv() @@ -112,7 +119,7 @@ def DeleteCard(): try: try: - logger.debug("Auth1")# + logger.debug("Auth1") desfire.authenticate(0x0, aes_master_key) except: logger.debug("Auth2") @@ -244,7 +251,7 @@ 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: diff --git a/app/services/settings.py b/app/services/settings.py index a60d589..2d42c98 100644 --- a/app/services/settings.py +++ b/app/services/settings.py @@ -1,5 +1,6 @@ -import os import logging +import os + logger = logging.getLogger(__name__) def verify_settings(): -- 2.51.2 From 4ce98a2be6a99739294b7b1c59d45fa8e62e0ac6 Mon Sep 17 00:00:00 2001 From: ahtlon Date: Wed, 29 Jul 2026 01:59:34 +0200 Subject: [PATCH 3/9] methinks I should've done this a lot sooner ran 'ruff format app' --- app/__init__.py | 2 +- app/controllers/aaManager.py | 77 ++++++++++++++++++----- app/controllers/cardManager.py | 66 ++++++++++++++++---- app/controllers/debugManager.py | 22 +++++-- app/controllers/doorManager.py | 7 ++- app/controllers/groupManager.py | 28 +++++++-- app/controllers/userManager.py | 50 ++++++++++++--- app/main.py | 16 ++--- app/model/models.py | 56 ++++++++++++++--- app/services/auth.py | 35 ++++++----- app/services/database.py | 6 +- app/services/door.py | 18 ++++-- app/services/scanner.py | 104 +++++++++++++++++++++----------- app/services/settings.py | 26 ++++---- 14 files changed, 377 insertions(+), 136 deletions(-) diff --git a/app/__init__.py b/app/__init__.py index 1217788..e4c3274 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -4,4 +4,4 @@ from .controllers import cardManager, userManager app = FastAPI() app.include_router(userManager.user_router) -app.include_router(cardManager.card_router) \ No newline at end of file +app.include_router(cardManager.card_router) diff --git a/app/controllers/aaManager.py b/app/controllers/aaManager.py index ad31e64..2a5f49c 100644 --- a/app/controllers/aaManager.py +++ b/app/controllers/aaManager.py @@ -12,40 +12,65 @@ from app.services.database import add_and_refresh, get_session aa_router = APIRouter(prefix="/api/v1/aa", tags=["AccessAuth"]) + @aa_router.post("/", response_model=AccessAuthorizationResponse) -def add_accessauth(*, db: Session = Depends(get_session), aa: AccessAuthorizationCreate, admin: bool = Depends(auth_is_admin)): +def add_accessauth( + *, + db: Session = Depends(get_session), + aa: AccessAuthorizationCreate, + admin: bool = Depends(auth_is_admin), +): logger.info(f"Creating accessauth with data: {aa}") if aa.timetables != []: timetables = [Timetable.model_validate(t) for t in aa.timetables] - else: timetables = [] + else: + timetables = [] if aa.oneshot is not None: oneshot = OneShotAccess.model_validate(aa.oneshot) - else: oneshot = None + else: + oneshot = None db_aa = AccessAuthorizationDB( name=aa.name, type=aa.type, is_active=aa.is_active, timetables=timetables, - oneshot=oneshot + oneshot=oneshot, ) return add_and_refresh(db, db_aa) + @aa_router.get("/", response_model=list[AccessAuthorizationResponse]) -def get_all_accessauths(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): +def get_all_accessauths( + db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin) +): return db.exec( - select(AccessAuthorizationDB) - .options(selectinload(AccessAuthorizationDB.timetables)) - ).all() + select(AccessAuthorizationDB).options( + selectinload(AccessAuthorizationDB.timetables) + ) + ).all() + @aa_router.get("/{aa_id}", response_model=AccessAuthorizationResponse) -def get_one_accessauth(*, db: Session = Depends(get_session), aa_id: int, admin: bool = Depends(auth_is_admin)): +def get_one_accessauth( + *, + db: Session = Depends(get_session), + aa_id: int, + admin: bool = Depends(auth_is_admin), +): db_aa = db.get(AccessAuthorizationDB, aa_id) if db_aa is None: raise HTTPException(status_code=404, detail="AA not found") return db_aa + @aa_router.put("/assign/{group_id}/{aa_id}", response_model=GroupResponse) -def assign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa_id: int, admin: bool = Depends(auth_is_admin)): +def assign_accessauth( + *, + db: Session = Depends(get_session), + group_id: int, + aa_id: int, + admin: bool = Depends(auth_is_admin), +): db_group = db.get(GroupDB, group_id) if db_group is None: raise HTTPException(status_code=404, detail="Group not found") @@ -53,12 +78,21 @@ def assign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa_i if db_aa is None: raise HTTPException(status_code=404, detail="AA not found") if db_aa in db_group.accessauths: - raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="AA already assigned to group") + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, detail="AA already assigned to group" + ) db_group.accessauths.append(db_aa) return add_and_refresh(db, db_group) + @aa_router.put("/unassign/{group_id}/{aa_id}", response_model=GroupResponse) -def unassign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa_id: int, admin: bool = Depends(auth_is_admin)): +def unassign_accessauth( + *, + db: Session = Depends(get_session), + group_id: int, + aa_id: int, + admin: bool = Depends(auth_is_admin), +): db_group = db.get(GroupDB, group_id) if db_group is None: raise HTTPException(status_code=404, detail="Group not found") @@ -70,8 +104,15 @@ def unassign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa db_group.accessauths.remove(db_aa) return add_and_refresh(db, db_group) + @aa_router.patch("/{aa_id}", response_model=AccessAuthorizationResponse) -def change_accessauth(*, db: Session = Depends(get_session), aa_id: int, aa: AccessAuthorizationUpdate, admin: bool = Depends(auth_is_admin)): +def change_accessauth( + *, + db: Session = Depends(get_session), + aa_id: int, + aa: AccessAuthorizationUpdate, + admin: bool = Depends(auth_is_admin), +): db_aa = db.get(AccessAuthorizationDB, aa_id) if db_aa is None: raise HTTPException(status_code=404, detail="AccessAuthorization not found") @@ -88,11 +129,17 @@ def change_accessauth(*, db: Session = Depends(get_session), aa_id: int, aa: Acc db_aa.sqlmodel_update(aa_data) return add_and_refresh(db, db_aa) + @aa_router.delete("/{aa_id}") -def delete_accessauth(*, db: Session = Depends(get_session), aa_id: int, admin: bool = Depends(auth_is_admin)): +def delete_accessauth( + *, + db: Session = Depends(get_session), + aa_id: int, + admin: bool = Depends(auth_is_admin), +): db_aa = db.get(AccessAuthorizationDB, aa_id) if db_aa is None: raise HTTPException(status_code=404, detail="AccessAuthorization not found") db.delete(db_aa) db.commit() - return {"message": "AccessAuthorization deleted successfully"} \ No newline at end of file + return {"message": "AccessAuthorization deleted successfully"} diff --git a/app/controllers/cardManager.py b/app/controllers/cardManager.py index 68cba9d..172c098 100644 --- a/app/controllers/cardManager.py +++ b/app/controllers/cardManager.py @@ -13,58 +13,100 @@ from app.services.scanner import DeleteCard, WriteNewCard card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"]) + def register_card(cardInput: CardCreate): key, uid = WriteNewCard() if key == None: logger.info("No card registered. Check logs!") - raise HTTPException(status.HTTP_417_EXPECTATION_FAILED, detail="No card registered. Check logs!") + raise HTTPException( + status.HTTP_417_EXPECTATION_FAILED, detail="No card registered. Check logs!" + ) card = Card( group_id=cardInput.group_id, key=key, name=cardInput.name, card_serial=uid, - enabled=cardInput.enabled - ) + enabled=cardInput.enabled, + ) return card + @card_router.post("/", response_model=Card) -def add_card(cardInput: CardCreate, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): - try: assert db.exec(select(Card).where(Card.name == cardInput.name)).one_or_none() == None +def add_card( + cardInput: CardCreate, + db: Session = Depends(get_session), + admin: bool = Depends(auth_is_admin), +): + try: + assert ( + db.exec(select(Card).where(Card.name == cardInput.name)).one_or_none() + == None + ) except AssertionError: raise HTTPException(status.HTTP_409_CONFLICT, detail="Name already used!") - try: assert db.exec(select(GroupDB).where(GroupDB.id == cardInput.group_id)).one_or_none() is not None + try: + assert ( + db.exec( + select(GroupDB).where(GroupDB.id == cardInput.group_id) + ).one_or_none() + is not None + ) except AssertionError: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="GroupID not found!") card = register_card(cardInput) return add_and_refresh(db, card) + @card_router.delete("/") -def del_card(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): +def del_card( + *, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin) +): key = DeleteCard() logger.info(key) try: card = db.exec(select(Card).where(Card.key == key)).one() except NoResultFound: logger.info(f"The key:'{key}' was not found in db!") - raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Key on card not found in DB. Please tell an admin about this. KEY={key}") + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Key on card not found in DB. Please tell an admin about this. KEY={key}", + ) db.delete(card) db.commit() return {"message": "Card deleted successfully"} + @card_router.get("/{group_id}", response_model=list[Card]) -def get_cards(*, db: Session = Depends(get_session), group_id: int, admin: bool = Depends(auth_is_admin)): +def get_cards( + *, + db: Session = Depends(get_session), + group_id: int, + admin: bool = Depends(auth_is_admin), +): cards = db.exec(select(Card).where(Card.group_id == group_id)).all() return cards + @card_router.patch("/{card_id}", response_model=Card) -def update_card(card_id: int, cardInput: CardUpdate, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): +def update_card( + card_id: int, + cardInput: CardUpdate, + db: Session = Depends(get_session), + admin: bool = Depends(auth_is_admin), +): db_card = db.get(Card, card_id) if db_card is None: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Card not found!") card_data = cardInput.model_dump(exclude_unset=True, exclude_none=True) if "group_id" in card_data: - try: assert db.exec(select(GroupDB).where(GroupDB.id == cardInput.group_id)).one_or_none() is not None + try: + assert ( + db.exec( + select(GroupDB).where(GroupDB.id == cardInput.group_id) + ).one_or_none() + is not None + ) except AssertionError: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="GroupID not found!") db_card.sqlmodel_update(card_data) - return add_and_refresh(db, db_card) \ No newline at end of file + return add_and_refresh(db, db_card) diff --git a/app/controllers/debugManager.py b/app/controllers/debugManager.py index 0f6addf..fd5b49a 100644 --- a/app/controllers/debugManager.py +++ b/app/controllers/debugManager.py @@ -15,20 +15,30 @@ debug_router = APIRouter( dependencies=[Depends(auth_is_admin)], ) + @debug_router.put("/addcard/") -def add_card_manually(groupid: int, card_key: str, name: str, enabled: bool, db: Session=Depends(get_session)): +def add_card_manually( + groupid: int, + card_key: str, + name: str, + enabled: bool, + db: Session = Depends(get_session), +): """Add cards manually (you also have to delete them manually)""" - logger.critical(f"Manual db change: adding a card with key: {card_key} to group: {groupid}") + logger.critical( + f"Manual db change: adding a card with key: {card_key} to group: {groupid}" + ) card = Card( group_id=groupid, key=card_key, name=name, enabled=enabled, - card_serial="00:00:00:00:00:00:00" - ) + card_serial="00:00:00:00:00:00:00", + ) add_and_refresh(db, card) return card + @debug_router.get("/rmcard/{card_id}") def remove_card_manually(card_id: str, db: Session = Depends(get_session)): try: @@ -39,7 +49,7 @@ def remove_card_manually(card_id: str, db: Session = Depends(get_session)): db.delete(card) db.commit() return {"message": "Card deleted successfully"} - + @debug_router.put("/getcards") def list_all_cards(db: Session = Depends(get_session)): @@ -49,4 +59,4 @@ def list_all_cards(db: Session = Depends(get_session)): out = [] for i in cards: out.append({i.key: i.group.name}) - return out \ No newline at end of file + return out diff --git a/app/controllers/doorManager.py b/app/controllers/doorManager.py index bfe22d0..88cc8c2 100644 --- a/app/controllers/doorManager.py +++ b/app/controllers/doorManager.py @@ -5,16 +5,19 @@ import app.services.door as doorService from app.services.auth import auth_is_admin from app.services.database import get_session -door_router = APIRouter(prefix="/api/v1/door",tags=["Door"]) +door_router = APIRouter(prefix="/api/v1/door", tags=["Door"]) + @door_router.put("/open") def open_door(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): doorService.opendoor() + @door_router.put("/close") def open_door(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): doorService.closedoor() + @door_router.post("/test") def test_access(input: str, db: Session = Depends(get_session)): - return doorService.checkAccess(input, db=db) \ No newline at end of file + return doorService.checkAccess(input, db=db) diff --git a/app/controllers/groupManager.py b/app/controllers/groupManager.py index 88a6786..0b76b51 100644 --- a/app/controllers/groupManager.py +++ b/app/controllers/groupManager.py @@ -1,4 +1,3 @@ - from fastapi import APIRouter, Depends, HTTPException, status from sqlmodel import Session, select @@ -8,24 +7,41 @@ from ..services.database import add_and_refresh, get_session group_router = APIRouter(prefix="/api/v1/groups", tags=["Group"]) + @group_router.get("/", response_model=list[GroupResponse]) -def get_groups(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): +def get_groups( + *, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin) +): groups = db.exec(select(GroupDB)).all() return groups + @group_router.post("/", response_model=GroupResponse) -def create_group(*, db: Session = Depends(get_session), group: GroupCreate, admin: bool = Depends(auth_is_admin)): +def create_group( + *, + db: Session = Depends(get_session), + group: GroupCreate, + admin: bool = Depends(auth_is_admin), +): db_group = GroupDB.model_validate(group) group = db.exec(select(GroupDB).where(GroupDB.name == db_group.name)).first() if group is not None: - raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Group already exists!") + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, detail="Group already exists!" + ) return add_and_refresh(db, db_group) + @group_router.delete("/{group_id}") -def delete_group(*, db: Session = Depends(get_session), group_id: int, admin: bool = Depends(auth_is_admin)): +def delete_group( + *, + db: Session = Depends(get_session), + group_id: int, + admin: bool = Depends(auth_is_admin), +): db_group = db.get(GroupDB, group_id) if db_group is None: raise HTTPException(status_code=404, detail="Group not found") db.delete(db_group) db.commit() - return {"message": "Group deleted successfully"} \ No newline at end of file + return {"message": "Group deleted successfully"} diff --git a/app/controllers/userManager.py b/app/controllers/userManager.py index f19a753..9a63813 100644 --- a/app/controllers/userManager.py +++ b/app/controllers/userManager.py @@ -12,33 +12,62 @@ from ..services.database import add_and_refresh, get_session user_router = APIRouter(tags=["Users"], prefix="/api/v1/users") + @user_router.post("/", response_model=UserResponse) -def create_user(*, db: Session = Depends(get_session), user: UserCreate, admin: bool = Depends(auth_is_admin)): +def create_user( + *, + db: Session = Depends(get_session), + user: UserCreate, + admin: bool = Depends(auth_is_admin), +): hashed_password = {"passwordhash": get_password_hash(user.password)} - try: assert db.exec(select(UserDB).where(UserDB.name == user.name)).one_or_none() == None + try: + assert ( + db.exec(select(UserDB).where(UserDB.name == user.name)).one_or_none() + == None + ) except AssertionError: raise HTTPException(status.HTTP_409_CONFLICT, detail="Name already used!") db_user = UserDB.model_validate(user, update=hashed_password) return add_and_refresh(db, db_user) + @user_router.get("/", response_model=list[UserResponse]) -def read_users(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): +def read_users( + *, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin) +): users = db.exec(select(UserDB)).all() return users + @user_router.get("/current", response_model=UserResponse) -def get_current_user(db: Session = Depends(get_session), user: UserDB = Depends(auth_user)): +def get_current_user( + db: Session = Depends(get_session), user: UserDB = Depends(auth_user) +): return user + @user_router.get("/{user_id}", response_model=UserResponse) -def read_user(*, db: Session = Depends(get_session), user_id: int, admin: bool = Depends(auth_is_admin)): +def read_user( + *, + db: Session = Depends(get_session), + user_id: int, + admin: bool = Depends(auth_is_admin), +): db_user = db.get(UserDB, user_id) if db_user is None: raise HTTPException(status_code=404, detail="User not found") return db_user + @user_router.patch("/{user_id}", response_model=UserResponse) -def update_user(*, db: Session = Depends(get_session), user_id: int, user: UserUpdate, admin: bool = Depends(auth_is_admin)): +def update_user( + *, + db: Session = Depends(get_session), + user_id: int, + user: UserUpdate, + admin: bool = Depends(auth_is_admin), +): db_user = db.get(UserDB, user_id) if db_user is None: raise HTTPException(status_code=404, detail="User not found") @@ -50,12 +79,17 @@ def update_user(*, db: Session = Depends(get_session), user_id: int, user: UserU db_user.sqlmodel_update(user_data, update=hashed_password) return add_and_refresh(db, db_user) + @user_router.delete("/{user_id}") -def delete_user(*, db: Session = Depends(get_session), user_id: int, admin: bool = Depends(auth_is_admin)): +def delete_user( + *, + db: Session = Depends(get_session), + user_id: int, + admin: bool = Depends(auth_is_admin), +): db_user = db.get(UserDB, user_id) if db_user is None: raise HTTPException(status_code=404, detail="User not found") db.delete(db_user) db.commit() return {"message": "User deleted successfully"} - diff --git a/app/main.py b/app/main.py index 3e6dde4..ee17ddf 100644 --- a/app/main.py +++ b/app/main.py @@ -28,6 +28,7 @@ oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") scanner = BackgroundScanner(db=get_db_session()) logging.basicConfig(level=logging.INFO) + @asynccontextmanager async def lifespan(app: FastAPI): verify_settings() @@ -37,17 +38,16 @@ async def lifespan(app: FastAPI): if not os.getenv("DISABLE_CARDS"): scanner.start() - logger.info("-"*63) + logger.info("-" * 63) logger.info("---- Documentation is at http://127.0.0.1:8000/api/v1/docs ----") - logger.info("-"*63) + logger.info("-" * 63) yield - #scanner.stop() + # scanner.stop() + app = FastAPI( - lifespan=lifespan, - docs_url="/api/v1/docs", - openapi_url="/api/v1/openapi.json" - ) + lifespan=lifespan, docs_url="/api/v1/docs", openapi_url="/api/v1/openapi.json" +) origins = [ "http://127.0.0.1", @@ -69,4 +69,4 @@ app.include_router(groupManager.group_router) app.include_router(cardManager.card_router) app.include_router(aaManager.aa_router) app.include_router(doorManager.door_router) -app.include_router(debugManager.debug_router) \ No newline at end of file +app.include_router(debugManager.debug_router) diff --git a/app/model/models.py b/app/model/models.py index 253887a..6935e13 100644 --- a/app/model/models.py +++ b/app/model/models.py @@ -8,89 +8,120 @@ from sqlmodel import Field, Relationship, SQLModel class Base(SQLModel): pass + #### User class UserBase(Base): name: str = Field(index=True, unique=True) email: str | None = None is_admin: bool = False + class UserResponse(UserBase): id: int + class UserCreate(UserBase): password: str + class UserDB(UserBase, table=True): id: int | None = Field(default=None, primary_key=True) passwordhash: str + class UserUpdate(Base): name: str | None = None email: str | None = None is_admin: bool | None = None password: str | None = None + #### Special class AaGroupLink(Base, table=True): - group_id: int | None = Field(default=None, foreign_key="groupdb.id", primary_key=True) - accessauth_id: int | None = Field(default=None, foreign_key="accessauthorizationdb.id", primary_key=True) + group_id: int | None = Field( + default=None, foreign_key="groupdb.id", primary_key=True + ) + accessauth_id: int | None = Field( + default=None, foreign_key="accessauthorizationdb.id", primary_key=True + ) + #### Token class Token(Base): access_token: str token_type: str + class TokenData(Base): username: str | None = None + #### Group class GroupBase(Base): name: str = Field(index=True, unique=True) + class GroupCreate(GroupBase): pass + class GroupDB(GroupBase, table=True): id: int | None = Field(default=None, primary_key=True) cards: list["Card"] = Relationship(back_populates="group") - accessauths: list["AccessAuthorizationDB"] = Relationship(back_populates="groups", link_model=AaGroupLink) + accessauths: list["AccessAuthorizationDB"] = Relationship( + back_populates="groups", link_model=AaGroupLink + ) + class GroupResponse(GroupBase): id: int cards: list["Card"] | None accessauths: list["AccessAuthorizationDB"] | None + #### AccessAuthorization class AccessAuthorizationBase(Base): name: str = Field(index=True) type: Literal["timetable", "oneshot", "somefuturespec"] is_active: bool + class AccessAuthorizationDB(AccessAuthorizationBase, table=True): id: int | None = Field(default=None, primary_key=True) type: str - groups: list["GroupDB"] = Relationship(back_populates="accessauths", link_model=AaGroupLink) - timetables: list["Timetable"] = Relationship(back_populates="accessauth", cascade_delete=True) - oneshot: "OneShotAccess" = Relationship(back_populates="accessauth", cascade_delete=True) + groups: list["GroupDB"] = Relationship( + back_populates="accessauths", link_model=AaGroupLink + ) + timetables: list["Timetable"] = Relationship( + back_populates="accessauth", cascade_delete=True + ) + oneshot: "OneShotAccess" = Relationship( + back_populates="accessauth", cascade_delete=True + ) + class OneShotAccessBase(Base): uses: int = 1 ends_at: datetime + class OneShotAccess(OneShotAccessBase, table=True): id: int | None = Field(default=None, primary_key=True) accessauth_id: int = Field(default=None, foreign_key="accessauthorizationdb.id") accessauth: AccessAuthorizationDB = Relationship(back_populates="oneshot") + class AccessAuthorizationCreate(AccessAuthorizationBase): timetables: list["TimetableCreate"] = [] oneshot: OneShotAccessBase | None = None - + @model_validator(mode="after") def check_type(self): if self.type == "timetable": if not self.timetables: - raise ValueError("timetable auths require at least one timetable object") + raise ValueError( + "timetable auths require at least one timetable object" + ) if self.oneshot is not None: raise ValueError("timetable auths are not allowed oneshot objects") elif self.type == "oneshot": @@ -102,19 +133,21 @@ class AccessAuthorizationCreate(AccessAuthorizationBase): raise ValueError("somefuturespec is not jet implemented. Please do not use") return self + class AccessAuthorizationResponse(AccessAuthorizationBase): id: int timetables: list["Timetable"] = [] oneshot: OneShotAccessBase | None = None groups: list["GroupDB"] + class AccessAuthorizationUpdate(Base): name: str | None = None type: Literal["timetable", "oneshot", "somefuturespec"] | None = None is_active: bool | None = None timetables: list["TimetableCreate"] | None = None oneshot: OneShotAccessBase | None = None - + #### Card class Card(Base, table=True): @@ -126,25 +159,30 @@ class Card(Base, table=True): group_id: int | None = Field(default=None, foreign_key="groupdb.id") group: GroupDB | None = Relationship(back_populates="cards") + class CardCreate(Base): name: str = Field(unique=True, max_length=32) enabled: bool = True group_id: int + class CardUpdate(Base): name: str | None = None enabled: bool | None = None group_id: int | None = None + class TimetableBase(Base): weekday: int = Field(le=6, ge=0) starttime: time duration: int = Field(gt=0, lt=1440) + class Timetable(TimetableBase, table=True): id: int | None = Field(default=None, primary_key=True) accessauth_id: int = Field(default=None, foreign_key="accessauthorizationdb.id") accessauth: AccessAuthorizationDB = Relationship(back_populates="timetables") + class TimetableCreate(TimetableBase): pass diff --git a/app/services/auth.py b/app/services/auth.py index 619c0d0..2cad156 100644 --- a/app/services/auth.py +++ b/app/services/auth.py @@ -17,7 +17,7 @@ from sqlmodel import Session, select from app.model.models import Token, TokenData, UserDB from app.services.database import * -SECRET_KEY = os.getenv("SECRET_KEY", default="ff"*16) +SECRET_KEY = os.getenv("SECRET_KEY", default="ff" * 16) ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 120 @@ -27,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: @@ -45,6 +49,7 @@ 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: @@ -55,14 +60,15 @@ def create_access_token(data: dict, expires_delta: timedelta | None = None): 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]) @@ -77,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}") @@ -108,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( @@ -123,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 \ No newline at end of file +def test_login(current_user: Annotated[UserDB, Depends(get_current_user)]) -> UserDB: + return current_user diff --git a/app/services/database.py b/app/services/database.py index 447fca3..9df4431 100644 --- a/app/services/database.py +++ b/app/services/database.py @@ -9,12 +9,14 @@ SQLALCHEMY_DATABASE_URL = getenv("SQLALCHEMY_DATABASE_URL", "sqlite:///./gatekee 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.") @@ -26,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 \ No newline at end of file + return obj diff --git a/app/services/door.py b/app/services/door.py index 93c508b..5b902fc 100644 --- a/app/services/door.py +++ b/app/services/door.py @@ -11,21 +11,25 @@ from app.services.database import Session, add_and_refresh 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") + def closeDoor(): global doorIsOpen doorIsOpen = False logger.info("Still needs gpio out") + def isDoorOpen(): return doorIsOpen + def decrementOneshot(db: Session, oneshot: OneShotAccess): data = oneshot.model_dump() if data["uses"] > 0: @@ -33,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()) @@ -43,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) @@ -61,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") - diff --git a/app/services/scanner.py b/app/services/scanner.py index 75e1fa1..89d48bf 100644 --- a/app/services/scanner.py +++ b/app/services/scanner.py @@ -30,11 +30,11 @@ from smartcard.Exceptions import CardRequestTimeoutException from app.services.door import checkAccess, openDoor -#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 @@ -43,10 +43,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.") + 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() @@ -55,51 +60,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) @@ -114,7 +126,7 @@ 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: @@ -127,11 +139,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) @@ -143,14 +157,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) @@ -166,8 +184,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 @@ -196,7 +214,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) @@ -205,18 +223,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( @@ -233,18 +261,22 @@ 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: @@ -299,7 +331,7 @@ 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: diff --git a/app/services/settings.py b/app/services/settings.py index 2d42c98..4a5a2bf 100644 --- a/app/services/settings.py +++ b/app/services/settings.py @@ -3,25 +3,27 @@ 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.") -- 2.51.2 From ce79d99a8cae0e362af8985696222e654a02afdb Mon Sep 17 00:00:00 2001 From: ahtlon Date: Wed, 29 Jul 2026 02:16:11 +0200 Subject: [PATCH 4/9] Vscode format on save maybe add a gitea action for this also --- .vscode/settings.json | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.vscode/settings.json b/.vscode/settings.json index b2b8866..cd1f90d 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -1,4 +1,8 @@ { + "[python]": { + "editor.formatOnSave": true, + "editor.defaultFormatter": "charliermarsh.ruff" + }, "python.testing.pytestArgs": [ "test" ], -- 2.51.2 From be4e6c8079f872b909adf190d1fab4d2fc88d7ea Mon Sep 17 00:00:00 2001 From: ahtlon Date: Wed, 29 Jul 2026 02:36:43 +0200 Subject: [PATCH 5/9] [ruff] do not format alembic folder --- pyproject.toml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index 9244d55..d6ec8a2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,3 +29,8 @@ python-desfire = ["poetry"] [tool.setuptools] py-modules = ["app"] + +[tool.ruff] +exclude = [ + "alembic" +] \ No newline at end of file -- 2.51.2 From 9cf21db9f9abf99e5ec952785b69b9d5de3cb7cc Mon Sep 17 00:00:00 2001 From: ahtlon Date: Wed, 29 Jul 2026 02:37:22 +0200 Subject: [PATCH 6/9] [lint] fix warnings not autofixed by ruff --- app/controllers/aaManager.py | 15 ++++++++++++--- app/controllers/cardManager.py | 9 ++++----- app/controllers/debugManager.py | 6 +++--- app/controllers/doorManager.py | 4 +++- app/controllers/userManager.py | 15 +++++++-------- app/services/auth.py | 6 +++--- app/services/database.py | 5 ++--- app/services/door.py | 6 +++--- app/services/scanner.py | 9 ++++----- 9 files changed, 41 insertions(+), 34 deletions(-) diff --git a/app/controllers/aaManager.py b/app/controllers/aaManager.py index 2a5f49c..4107b52 100644 --- a/app/controllers/aaManager.py +++ b/app/controllers/aaManager.py @@ -1,15 +1,24 @@ import logging -logger = logging.getLogger(__name__) - from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import selectinload from sqlmodel import Session, select -from app.model.models import * +from app.model.models import ( + AccessAuthorizationCreate, + AccessAuthorizationResponse, + Timetable, + AccessAuthorizationDB, + OneShotAccess, + GroupResponse, + GroupDB, + AccessAuthorizationUpdate, +) from app.services.auth import auth_is_admin from app.services.database import add_and_refresh, get_session +logger = logging.getLogger(__name__) + aa_router = APIRouter(prefix="/api/v1/aa", tags=["AccessAuth"]) diff --git a/app/controllers/cardManager.py b/app/controllers/cardManager.py index 172c098..4506349 100644 --- a/app/controllers/cardManager.py +++ b/app/controllers/cardManager.py @@ -1,7 +1,4 @@ import logging - -logger = logging.getLogger(__name__) - from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.exc import NoResultFound from sqlmodel import Session, select @@ -11,12 +8,14 @@ from app.services.auth import auth_is_admin from app.services.database import add_and_refresh, get_session from app.services.scanner import DeleteCard, WriteNewCard +logger = logging.getLogger(__name__) + card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"]) def register_card(cardInput: CardCreate): key, uid = WriteNewCard() - if key == None: + if key is None: logger.info("No card registered. Check logs!") raise HTTPException( status.HTTP_417_EXPECTATION_FAILED, detail="No card registered. Check logs!" @@ -40,7 +39,7 @@ def add_card( try: assert ( db.exec(select(Card).where(Card.name == cardInput.name)).one_or_none() - == None + is None ) except AssertionError: raise HTTPException(status.HTTP_409_CONFLICT, detail="Name already used!") diff --git a/app/controllers/debugManager.py b/app/controllers/debugManager.py index fd5b49a..fd4b507 100644 --- a/app/controllers/debugManager.py +++ b/app/controllers/debugManager.py @@ -1,14 +1,14 @@ import logging - -logger = logging.getLogger(__name__) from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import exc from sqlmodel import Session, select -from app.model.models import * +from app.model.models import Card from app.services.auth import auth_is_admin from app.services.database import add_and_refresh, get_session +logger = logging.getLogger(__name__) + debug_router = APIRouter( prefix="/api/v1/debug", tags=["Debug items - maybe dont show this in UI"], diff --git a/app/controllers/doorManager.py b/app/controllers/doorManager.py index 88cc8c2..449b964 100644 --- a/app/controllers/doorManager.py +++ b/app/controllers/doorManager.py @@ -14,7 +14,9 @@ def open_door(db: Session = Depends(get_session), admin: bool = Depends(auth_is_ @door_router.put("/close") -def open_door(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): +def close_door( + db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin) +): doorService.closedoor() diff --git a/app/controllers/userManager.py b/app/controllers/userManager.py index 9a63813..130dfe1 100644 --- a/app/controllers/userManager.py +++ b/app/controllers/userManager.py @@ -1,14 +1,13 @@ import logging - -logger = logging.getLogger(__name__) - from fastapi import APIRouter, Depends, HTTPException, status from sqlmodel import Session, select -from ..model.models import UserCreate, UserDB, UserResponse, UserUpdate -from ..services.auth import auth_is_admin, get_password_hash -from ..services.auth import get_current_user as auth_user -from ..services.database import add_and_refresh, get_session +from app.model.models import UserCreate, UserDB, UserResponse, UserUpdate +from app.services.auth import auth_is_admin, get_password_hash +from app.services.auth import get_current_user as auth_user +from app.services.database import add_and_refresh, get_session + +logger = logging.getLogger(__name__) user_router = APIRouter(tags=["Users"], prefix="/api/v1/users") @@ -24,7 +23,7 @@ def create_user( try: assert ( db.exec(select(UserDB).where(UserDB.name == user.name)).one_or_none() - == None + is None ) except AssertionError: raise HTTPException(status.HTTP_409_CONFLICT, detail="Name already used!") diff --git a/app/services/auth.py b/app/services/auth.py index 2cad156..5b54232 100644 --- a/app/services/auth.py +++ b/app/services/auth.py @@ -1,6 +1,4 @@ import logging - -logger = logging.getLogger(__name__) import os import secrets import string @@ -15,7 +13,9 @@ from pwdlib import PasswordHash from sqlmodel import Session, select from app.model.models import Token, TokenData, UserDB -from app.services.database import * +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" diff --git a/app/services/database.py b/app/services/database.py index 9df4431..b2dd2e6 100644 --- a/app/services/database.py +++ b/app/services/database.py @@ -1,9 +1,8 @@ import logging +from os import getenv, path +from sqlmodel import Session, SQLModel, create_engine logger = logging.getLogger(__name__) -from os import getenv, path - -from sqlmodel import Session, SQLModel, create_engine SQLALCHEMY_DATABASE_URL = getenv("SQLALCHEMY_DATABASE_URL", "sqlite:///./gatekeeper.db") diff --git a/app/services/door.py b/app/services/door.py index 5b902fc..c734ee9 100644 --- a/app/services/door.py +++ b/app/services/door.py @@ -1,14 +1,14 @@ import logging - -logger = logging.getLogger(__name__) from datetime import date, datetime, timedelta from sqlalchemy import exc from sqlmodel import select -from app.model.models import * +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 diff --git a/app/services/scanner.py b/app/services/scanner.py index 89d48bf..72e056c 100644 --- a/app/services/scanner.py +++ b/app/services/scanner.py @@ -1,7 +1,4 @@ import logging - -logger = logging.getLogger(__name__) - import os import secrets import threading @@ -30,6 +27,8 @@ from smartcard.Exceptions import CardRequestTimeoutException from app.services.door import checkAccess, openDoor +logger = logging.getLogger(__name__) + # ENV vars load_dotenv() MIFARE_APP_MASTER_KEY = os.getenv("MIFARE_APP_MASTER_KEY") @@ -45,7 +44,7 @@ MIFARE_ENCRYPTED_FILE_ID = 0x1 def checkForKey(): - if MIFARE_APP_MASTER_KEY == None: + if MIFARE_APP_MASTER_KEY is None: logger.critical("NO MASTER KEY LOADED") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -334,7 +333,7 @@ class BackgroundScanner: def _check_db(self, key): check = checkAccess(key, self.db) - if check == True: + if check: openDoor() logger.info("Access granted!") else: -- 2.51.2 From 506076cc1ee64deb9db39073233006f87b1a76c2 Mon Sep 17 00:00:00 2001 From: ahtlon Date: Wed, 29 Jul 2026 02:38:14 +0200 Subject: [PATCH 7/9] [lint] sort imports --- app/controllers/aaManager.py | 10 +++++----- app/controllers/cardManager.py | 1 + app/controllers/debugManager.py | 1 + app/controllers/userManager.py | 1 + app/services/database.py | 1 + 5 files changed, 9 insertions(+), 5 deletions(-) diff --git a/app/controllers/aaManager.py b/app/controllers/aaManager.py index 4107b52..48a133b 100644 --- a/app/controllers/aaManager.py +++ b/app/controllers/aaManager.py @@ -6,13 +6,13 @@ from sqlmodel import Session, select from app.model.models import ( AccessAuthorizationCreate, - AccessAuthorizationResponse, - Timetable, AccessAuthorizationDB, - OneShotAccess, - GroupResponse, - GroupDB, + AccessAuthorizationResponse, AccessAuthorizationUpdate, + GroupDB, + GroupResponse, + OneShotAccess, + Timetable, ) from app.services.auth import auth_is_admin from app.services.database import add_and_refresh, get_session diff --git a/app/controllers/cardManager.py b/app/controllers/cardManager.py index 4506349..c5ca4af 100644 --- a/app/controllers/cardManager.py +++ b/app/controllers/cardManager.py @@ -1,4 +1,5 @@ import logging + from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.exc import NoResultFound from sqlmodel import Session, select diff --git a/app/controllers/debugManager.py b/app/controllers/debugManager.py index fd4b507..0cbeafa 100644 --- a/app/controllers/debugManager.py +++ b/app/controllers/debugManager.py @@ -1,4 +1,5 @@ import logging + from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import exc from sqlmodel import Session, select diff --git a/app/controllers/userManager.py b/app/controllers/userManager.py index 130dfe1..4185a2b 100644 --- a/app/controllers/userManager.py +++ b/app/controllers/userManager.py @@ -1,4 +1,5 @@ import logging + from fastapi import APIRouter, Depends, HTTPException, status from sqlmodel import Session, select diff --git a/app/services/database.py b/app/services/database.py index b2dd2e6..68ef002 100644 --- a/app/services/database.py +++ b/app/services/database.py @@ -1,5 +1,6 @@ import logging from os import getenv, path + from sqlmodel import Session, SQLModel, create_engine logger = logging.getLogger(__name__) -- 2.51.2 From 8aa17c1b84949b74caf288a05e0afd4f792cfe1f Mon Sep 17 00:00:00 2001 From: ahtlon Date: Wed, 29 Jul 2026 02:40:46 +0200 Subject: [PATCH 8/9] [lint] lint + format tests --- test.py | 41 +++++++++---- test/conftest.py | 61 ++++++++++--------- test/test_main.py | 2 + test/test_models.py | 28 +++++---- test/test_services/test_aa_manager.py | 75 +++++++++++------------- test/test_services/test_auth.py | 37 +++++++----- test/test_services/test_card_manager.py | 33 ++++------- test/test_services/test_database.py | 9 +-- test/test_services/test_door.py | 47 ++++++++++++--- test/test_services/test_group_manager.py | 8 +-- test/test_services/test_user_manager.py | 39 +++++------- 11 files changed, 208 insertions(+), 172 deletions(-) diff --git a/test.py b/test.py index 17c4a87..943726b 100644 --- a/test.py +++ b/test.py @@ -15,21 +15,32 @@ It performs the following steps: import logging import os +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 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 - -from dotenv import load_dotenv # Please make sure to yet your own keys here before running this script 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 @@ -118,8 +129,12 @@ desfire.authenticate(0x0, aes_app_mk) # Change file read and write keys (diversified) 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) +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 +) print("Changing file read key...") aes_file_read_key = DESFireKey(aes_keysettings, read_div_key_bytes) @@ -154,11 +169,13 @@ print(" - File created successfully.") print("Writing UID to encrypted file...") data = [0x0] + uid assert len(data) == 8 -desfire.write_file_data(MIFARE_ENCRYPTED_FILE_ID, 0x0, file_data.encryption, get_list(data)) +desfire.write_file_data( + MIFARE_ENCRYPTED_FILE_ID, 0x0, file_data.encryption, get_list(data) +) print("Reading from encrypted file...") rdata = desfire.read_file_data(MIFARE_ENCRYPTED_FILE_ID, file_data) assert rdata == data print(" - Data written successfully.") -print("Personalization finished.") \ No newline at end of file +print("Personalization finished.") diff --git a/test/conftest.py b/test/conftest.py index a60fa9d..1a4bf40 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,18 +1,29 @@ -import pytest -from fastapi.testclient import TestClient -from sqlmodel import Session, create_engine, SQLModel -from sqlalchemy.orm import sessionmaker -from sqlalchemy.pool import StaticPool from datetime import time +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.pool import StaticPool +from sqlmodel import Session, SQLModel, create_engine + from app.main import app -from app.model.models import UserDB, Card, GroupDB, AccessAuthorizationDB, Timetable, AaGroupLink +from app.model.models import ( + AccessAuthorizationDB, + Card, + GroupDB, + Timetable, + UserDB, +) from app.services.database import get_session # Use in-memory SQLite for testing TEST_SQLALCHEMY_DATABASE_URL = "sqlite://" -engine = create_engine(TEST_SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}, poolclass=StaticPool) +engine = create_engine( + TEST_SQLALCHEMY_DATABASE_URL, + connect_args={"check_same_thread": False}, + poolclass=StaticPool, +) + @pytest.fixture(scope="function") def db_session(): @@ -26,6 +37,7 @@ def db_session(): @pytest.fixture(scope="function") def client(db_session): """Create a test client with a database session override.""" + def override_get_session(): yield db_session @@ -39,10 +51,9 @@ def client(db_session): def admin_user(db_session): """Create an admin user for testing.""" from app.services.auth import get_password_hash + admin = UserDB( - name="admin", - passwordhash=get_password_hash("admin123"), - is_admin=True + name="admin", passwordhash=get_password_hash("admin123"), is_admin=True ) db_session.add(admin) db_session.commit() @@ -54,10 +65,9 @@ def admin_user(db_session): def regular_user(db_session): """Create a regular user for testing.""" from app.services.auth import get_password_hash + user = UserDB( - name="user", - passwordhash=get_password_hash("user123"), - is_admin=False + name="user", passwordhash=get_password_hash("user123"), is_admin=False ) db_session.add(user) db_session.commit() @@ -69,8 +79,7 @@ def regular_user(db_session): def auth_headers(client, admin_user): """Get authentication headers for admin user.""" response = client.post( - "/api/v1/token", - data={"username": admin_user.name, "password": "admin123"} + "/api/v1/token", data={"username": admin_user.name, "password": "admin123"} ) token = response.json()["access_token"] return {"Authorization": f"Bearer {token}"} @@ -80,8 +89,7 @@ def auth_headers(client, admin_user): def user_auth_headers(client, regular_user): """Get authentication headers for regular user.""" response = client.post( - "/api/v1/token", - data={"username": regular_user.name, "password": "user123"} + "/api/v1/token", data={"username": regular_user.name, "password": "user123"} ) token = response.json()["access_token"] return {"Authorization": f"Bearer {token}"} @@ -100,7 +108,13 @@ def test_group(db_session): @pytest.fixture def test_card(db_session, test_group): """Create a test card.""" - card = Card(key="test-key-123", group_id=test_group.id, enabled=True, name="test_card", card_serial="00:00:00:00:00:00:00") + card = Card( + key="test-key-123", + group_id=test_group.id, + enabled=True, + name="test_card", + card_serial="00:00:00:00:00:00:00", + ) db_session.add(card) db_session.commit() db_session.refresh(card) @@ -110,16 +124,9 @@ def test_card(db_session, test_group): @pytest.fixture def test_aa_tt(db_session): """Create a test access authorization with timetable.""" - tt = Timetable( - weekday=1, - starttime=time(1, 0, 0, 0), - duration=50 - ) + tt = Timetable(weekday=1, starttime=time(1, 0, 0, 0), duration=50) aa = AccessAuthorizationDB( - name="Test AA", - is_active=True, - type="timetable", - timetables=[tt] + name="Test AA", is_active=True, type="timetable", timetables=[tt] ) db_session.add(aa) db_session.commit() diff --git a/test/test_main.py b/test/test_main.py index daa08cb..204f294 100644 --- a/test/test_main.py +++ b/test/test_main.py @@ -4,9 +4,11 @@ def test_app_startup(client): # Application should respond (even if it's a 404) assert response.status_code in [404, 200] + def test_router_includes(): """Test that all routers are included in the app.""" from app.main import app + routes = [route.path for route in app.routes] # Check that router prefixes are present diff --git a/test/test_models.py b/test/test_models.py index f30710f..59b3deb 100644 --- a/test/test_models.py +++ b/test/test_models.py @@ -1,11 +1,18 @@ -import pytest import datetime + from app.model.models import ( - UserBase, UserResponse, UserCreate, UserDB, UserUpdate, - GroupBase, GroupCreate, GroupDB, GroupResponse, - AccessAuthorizationBase, AccessAuthorizationCreate, - AccessAuthorizationDB, AccessAuthorizationResponse, AccessAuthorizationUpdate, - Card, Timetable, TimetableCreate, Token, TokenData, AaGroupLink + AaGroupLink, + AccessAuthorizationBase, + AccessAuthorizationCreate, + Card, + GroupBase, + GroupCreate, + TimetableCreate, + Token, + TokenData, + UserBase, + UserCreate, + UserUpdate, ) @@ -18,7 +25,9 @@ def test_user_models(): assert user_base.is_admin is False # Test UserCreate - user_create = UserCreate(name="New User", email="new@example.com", password="secret123") + user_create = UserCreate( + name="New User", email="new@example.com", password="secret123" + ) assert user_create.password == "secret123" # Test UserUpdate @@ -48,10 +57,7 @@ def test_access_authorization_models(): # Test AccessAuthorizationCreate with timetables timetable_create = TimetableCreate(weekday=1, starttime="08:00", duration=60) aa_create = AccessAuthorizationCreate( - name="New AA", - is_active=False, - type="timetable", - timetables=[timetable_create] + name="New AA", is_active=False, type="timetable", timetables=[timetable_create] ) assert aa_create.name == "New AA" assert aa_create.is_active is False diff --git a/test/test_services/test_aa_manager.py b/test/test_services/test_aa_manager.py index 1714257..fdfce11 100644 --- a/test/test_services/test_aa_manager.py +++ b/test/test_services/test_aa_manager.py @@ -1,7 +1,3 @@ -import pytest -from fastapi import status - - def test_create_access_auth_tt(client, auth_headers): """Test creating a new access authorization.""" aa_data = { @@ -10,8 +6,8 @@ def test_create_access_auth_tt(client, auth_headers): "is_active": True, "timetables": [ {"weekday": 1, "starttime": "08:00", "duration": 60}, - {"weekday": 2, "starttime": "09:00", "duration": 90} - ] + {"weekday": 2, "starttime": "09:00", "duration": 90}, + ], } response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers) @@ -24,16 +20,14 @@ def test_create_access_auth_tt(client, auth_headers): assert "id" in data assert len(data["timetables"]) == 2 + def test_create_access_auth_os(client, auth_headers): """Test creating a new access authorization with oneshot type.""" aa_data = { "name": "New os_AA", "type": "oneshot", "is_active": True, - "oneshot": { - "uses": 1, - "ends_at": "2029-07-27" - } + "oneshot": {"uses": 1, "ends_at": "2029-07-27"}, } response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers) @@ -46,6 +40,7 @@ def test_create_access_auth_os(client, auth_headers): assert "id" in data assert data["oneshot"]["uses"] == 1 + def test_create_wrong_aa_type(client, auth_headers): """Test creating a new access authorization with oneshot type.""" aa_data = { @@ -57,6 +52,7 @@ def test_create_wrong_aa_type(client, auth_headers): response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers) assert response.status_code == 422 + def test_get_all_access_auths(client, auth_headers, test_aa_tt): """Test retrieving all access authorizations.""" response = client.get("/api/v1/aa/", headers=auth_headers) @@ -88,8 +84,7 @@ def test_get_nonexistent_access_auth(client, auth_headers): def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa_tt): """Test assigning an access authorization to a group.""" response = client.put( - f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", - headers=auth_headers + f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers ) assert response.status_code == 200 @@ -99,15 +94,18 @@ def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa_t # Note: The response model might not include the full relationship -def test_assign_already_assigned_access_auth(client, auth_headers, test_group, test_aa_tt): +def test_assign_already_assigned_access_auth( + client, auth_headers, test_group, test_aa_tt +): """Test assigning an already assigned access authorization.""" # First assignment - client.put(f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers) + client.put( + f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers + ) # Second assignment should indicate it's already assigned response = client.put( - f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", - headers=auth_headers + f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers ) # According to the code, this returns 409 with "already assigned" message assert response.status_code == 409 @@ -117,12 +115,13 @@ def test_assign_already_assigned_access_auth(client, auth_headers, test_group, t def test_unassign_access_auth_from_group(client, auth_headers, test_group, test_aa_tt): """Test unassigning an access authorization from a group.""" # First assign - client.put(f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers) + client.put( + f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers + ) # Then unassign response = client.put( - f"/api/v1/aa/unassign/{test_group.id}/{test_aa_tt.id}", - headers=auth_headers + f"/api/v1/aa/unassign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers ) assert response.status_code == 200 @@ -130,35 +129,33 @@ def test_unassign_access_auth_from_group(client, auth_headers, test_group, test_ def test_unassign_nonexistent_assignment(client, auth_headers, test_group, test_aa_tt): """Test unassigning a non-existent assignment.""" response = client.put( - f"/api/v1/aa/unassign/{test_group.id}/{test_aa_tt.id}", - headers=auth_headers + f"/api/v1/aa/unassign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers ) assert response.status_code == 404 def test_assign_to_nonexistent_group(client, auth_headers, test_aa_tt): """Test assigning an AA to a non-existent group.""" - response = client.put(f"/api/v1/aa/assign/99999/{test_aa_tt.id}", headers=auth_headers) + response = client.put( + f"/api/v1/aa/assign/99999/{test_aa_tt.id}", headers=auth_headers + ) assert response.status_code == 404 def test_assign_nonexistent_aa(client, auth_headers, test_group): """Test assigning a non-existent AA to a group.""" - response = client.put(f"/api/v1/aa/assign/{test_group.id}/99999", headers=auth_headers) + response = client.put( + f"/api/v1/aa/assign/{test_group.id}/99999", headers=auth_headers + ) assert response.status_code == 404 def test_update_access_auth(client, auth_headers, test_aa_tt): """Test updating an access authorization.""" - update_data = { - "name": "Updated AA", - "is_active": False - } + update_data = {"name": "Updated AA", "is_active": False} response = client.patch( - f"/api/v1/aa/{test_aa_tt.id}", - json=update_data, - headers=auth_headers + f"/api/v1/aa/{test_aa_tt.id}", json=update_data, headers=auth_headers ) assert response.status_code == 200 @@ -170,15 +167,11 @@ def test_update_access_auth(client, auth_headers, test_aa_tt): def test_update_access_auth_with_timetables(client, auth_headers, test_aa_tt): """Test updating an access authorization with new timetables.""" update_data = { - "timetables": [ - {"weekday": 5, "starttime": "10:00", "duration": 120} - ] + "timetables": [{"weekday": 5, "starttime": "10:00", "duration": 120}] } response = client.patch( - f"/api/v1/aa/{test_aa_tt.id}", - json=update_data, - headers=auth_headers + f"/api/v1/aa/{test_aa_tt.id}", json=update_data, headers=auth_headers ) assert response.status_code == 200 jresponse = response.json() @@ -216,9 +209,9 @@ def test_aa_tt_operations_by_non_admin(client, test_aa_tt, user_auth_headers): """Test that non-admin users cannot perform AA operations.""" # Try to create an AA response = client.post( - "/api/v1/aa/", - json={"name": "test", "is_active": True, "timetables": []}, - headers=user_auth_headers + "/api/v1/aa/", + json={"name": "test", "is_active": True, "timetables": []}, + headers=user_auth_headers, ) assert response.status_code == 403 @@ -227,5 +220,7 @@ def test_aa_tt_operations_by_non_admin(client, test_aa_tt, user_auth_headers): assert response.status_code == 403 # Try to assign AA - response = client.put(f"/api/v1/aa/assign/1/{test_aa_tt.id}", headers=user_auth_headers) + response = client.put( + f"/api/v1/aa/assign/1/{test_aa_tt.id}", headers=user_auth_headers + ) assert response.status_code == 403 diff --git a/test/test_services/test_auth.py b/test/test_services/test_auth.py index 20fcd70..8a5d427 100644 --- a/test/test_services/test_auth.py +++ b/test/test_services/test_auth.py @@ -1,12 +1,19 @@ +from datetime import timedelta + import pytest -from datetime import datetime, timedelta, timezone from fastapi import HTTPException, status -from app.services.auth import ( - verify_password, get_password_hash, get_user, authenticate_user, - create_access_token, get_current_user, auth_is_admin, create_first_user -) + from app.model.models import UserDB -from jwt.exceptions import InvalidTokenError +from app.services.auth import ( + auth_is_admin, + authenticate_user, + create_access_token, + create_first_user, + get_current_user, + get_password_hash, + get_user, + verify_password, +) def test_password_hashing(): @@ -84,7 +91,7 @@ def test_create_access_token(): def test_get_current_user(db_session, admin_user): """Test getting current user from token.""" - from app.services.auth import create_access_token, get_current_user + from app.services.auth import create_access_token # Create token for admin user token = create_access_token(data={"sub": admin_user.name}) @@ -102,7 +109,9 @@ def test_get_current_user(db_session, admin_user): # Test expired token (create token with past expiration) past_expire = timedelta(minutes=-100) - expired_token = create_access_token(data={"sub": admin_user.name}, expires_delta=past_expire) + expired_token = create_access_token( + data={"sub": admin_user.name}, expires_delta=past_expire + ) with pytest.raises(HTTPException) as exc_info: get_current_user(token=expired_token) @@ -111,7 +120,7 @@ def test_get_current_user(db_session, admin_user): def test_auth_is_admin(db_session, admin_user, regular_user): """Test admin authorization check.""" - from app.services.auth import create_access_token, auth_is_admin + from app.services.auth import create_access_token # Create token for admin user admin_token = create_access_token(data={"sub": admin_user.name}) @@ -133,6 +142,7 @@ def test_create_first_user(db_session): """Test automatic creation of first admin user.""" # Clear any existing users from sqlmodel import select + db_session.exec(select(UserDB)).all() for user in db_session.exec(select(UserDB)).all(): db_session.delete(user) @@ -158,8 +168,7 @@ def test_token_endpoint(client, admin_user): """Test the token endpoint for login.""" # Test successful login response = client.post( - "/api/v1/token", - data={"username": admin_user.name, "password": "admin123"} + "/api/v1/token", data={"username": admin_user.name, "password": "admin123"} ) assert response.status_code == 200 data = response.json() @@ -168,15 +177,13 @@ def test_token_endpoint(client, admin_user): # Test failed login with wrong password response = client.post( - "/api/v1/token", - data={"username": admin_user.name, "password": "wrongpassword"} + "/api/v1/token", data={"username": admin_user.name, "password": "wrongpassword"} ) assert response.status_code == 401 # Test failed login with non-existent user response = client.post( - "/api/v1/token", - data={"username": "nonexistent", "password": "password"} + "/api/v1/token", data={"username": "nonexistent", "password": "password"} ) assert response.status_code == 401 diff --git a/test/test_services/test_card_manager.py b/test/test_services/test_card_manager.py index 2eca66d..b76ec14 100644 --- a/test/test_services/test_card_manager.py +++ b/test/test_services/test_card_manager.py @@ -1,6 +1,3 @@ -import pytest -from fastapi import status - def test_get_cards_for_group(client, auth_headers, test_group, test_card): """Test getting all cards for a group.""" response = client.get(f"/api/v1/cards/{test_group.id}", headers=auth_headers) @@ -23,24 +20,20 @@ def test_get_cards_for_nonexistent_group(client, auth_headers): def test_card_operations_by_non_admin(client, test_group, user_auth_headers): """Test that non-admin users cannot perform card operations.""" # Try to add a card - response = client.post(f"/api/v1/cards/", headers=user_auth_headers) + response = client.post("/api/v1/cards/", headers=user_auth_headers) assert response.status_code == 403 # Try to get cards response = client.get(f"/api/v1/cards/{test_group.id}", headers=user_auth_headers) assert response.status_code == 403 + def test_update_card(client, auth_headers, test_group, test_card): """Test Patching a card entity""" - update_data = { - "name": "changed_name", - "enabled": "False" - } + update_data = {"name": "changed_name", "enabled": "False"} response = client.patch( - f"/api/v1/cards/{test_card.id}", - json=update_data, - headers=auth_headers + f"/api/v1/cards/{test_card.id}", json=update_data, headers=auth_headers ) assert response.status_code == 200 @@ -49,27 +42,21 @@ def test_update_card(client, auth_headers, test_group, test_card): assert data["enabled"] == False assert data["group_id"] == test_card.group_id + def test_update_wrong_card(client, auth_headers, test_group, test_card): """Test Patching a card entity""" - response = client.patch( - f"/api/v1/cards/9999", - json={}, - headers=auth_headers - ) + response = client.patch("/api/v1/cards/9999", json={}, headers=auth_headers) assert response.status_code == 404 assert "Card not found" in response.json()["detail"] + def test_update_card_with_wrong_group(client, auth_headers, test_group, test_card): """Test Patching a card entity with wrong group""" - update_data = { - "group_id": "9999" - } + update_data = {"group_id": "9999"} response = client.patch( - f"/api/v1/cards/{test_card.id}", - json=update_data, - headers=auth_headers + f"/api/v1/cards/{test_card.id}", json=update_data, headers=auth_headers ) assert response.status_code == 404 - assert "GroupID not found" in response.json()["detail"] \ No newline at end of file + assert "GroupID not found" in response.json()["detail"] diff --git a/test/test_services/test_database.py b/test/test_services/test_database.py index c4a305d..5030922 100644 --- a/test/test_services/test_database.py +++ b/test/test_services/test_database.py @@ -1,13 +1,14 @@ -import pytest -from sqlmodel import Session, select -from app.services.database import create_db_and_tables, get_session, add_and_refresh -from app.model.models import UserDB, GroupDB, Card +from sqlmodel import Session + +from app.model.models import UserDB +from app.services.database import add_and_refresh, create_db_and_tables def test_create_db_and_tables(): """Test database and tables creation.""" # This is primarily an integration test from sqlalchemy import inspect + from app.services.database import engine create_db_and_tables() diff --git a/test/test_services/test_door.py b/test/test_services/test_door.py index 7e332e8..e733738 100644 --- a/test/test_services/test_door.py +++ b/test/test_services/test_door.py @@ -1,7 +1,16 @@ -import pytest import datetime + +import pytest + +from app.model.models import ( + AccessAuthorizationDB, + Card, + GroupDB, + OneShotAccess, + Timetable, +) from app.services.door import checkAccess -from app.model.models import Card, GroupDB, AccessAuthorizationDB, Timetable, OneShotAccess + def test_check_access_with_valid_timetable(db_session): # Setup: create card with valid access @@ -9,13 +18,19 @@ def test_check_access_with_valid_timetable(db_session): db_session.add(group) db_session.commit() - card = Card(key="test-key-123", group_id=group.id, enabled=True, name="test_card", card_serial="00:00:00:00:00:00:00") + card = Card( + key="test-key-123", + group_id=group.id, + enabled=True, + name="test_card", + card_serial="00:00:00:00:00:00:00", + ) db_session.add(card) timetable = Timetable( weekday=datetime.datetime.weekday(datetime.date.today()), starttime=datetime.datetime.now().time(), - duration=120 # 2 hours + duration=120, # 2 hours ) db_session.add(timetable) @@ -30,19 +45,26 @@ def test_check_access_with_valid_timetable(db_session): result = checkAccess("test-key-123", db_session) assert result == True + def test_check_access_outside_hours(db_session): # Test when current time is outside valid hours group = GroupDB(name="Test Group") db_session.add(group) db_session.commit() - card = Card(key="test-key-123", group_id=group.id, enabled=True, name="test_card", card_serial="00:00:00:00:00:00:00") + card = Card( + key="test-key-123", + group_id=group.id, + enabled=True, + name="test_card", + card_serial="00:00:00:00:00:00:00", + ) db_session.add(card) timetable = Timetable( weekday=datetime.datetime.weekday(datetime.date.today()), starttime=datetime.time(1, 0), - duration=1 # 2 hours + duration=1, # 2 hours ) db_session.add(timetable) @@ -55,18 +77,24 @@ def test_check_access_outside_hours(db_session): result = checkAccess("test-key-123", db_session) assert result == False + def test_check_access_with_valid_oneshot(db_session): # Setup: create card with valid access group = GroupDB(name="Test Group") db_session.add(group) db_session.commit() - card = Card(key="test-key-123", group_id=group.id, enabled=True, name="test_card", card_serial="00:00:00:00:00:00:00") + card = Card( + key="test-key-123", + group_id=group.id, + enabled=True, + name="test_card", + card_serial="00:00:00:00:00:00:00", + ) db_session.add(card) oneshot = OneShotAccess( - uses=1, - ends_at=datetime.datetime.now() + datetime.timedelta(days=1) + uses=1, ends_at=datetime.datetime.now() + datetime.timedelta(days=1) ) db_session.add(oneshot) @@ -82,6 +110,7 @@ def test_check_access_with_valid_oneshot(db_session): assert result == True assert aa.oneshot.uses == 0 + def test_check_access_invalid_card(db_session): # Should raise exception for non-existent card with pytest.raises(Exception): diff --git a/test/test_services/test_group_manager.py b/test/test_services/test_group_manager.py index 6eb3dbf..f9fa39e 100644 --- a/test/test_services/test_group_manager.py +++ b/test/test_services/test_group_manager.py @@ -1,7 +1,3 @@ -import pytest -from fastapi import status - - def test_create_group(client, auth_headers): """Test creating a new group.""" group_data = {"name": "New Test Group"} @@ -57,9 +53,7 @@ def test_group_operations_by_non_admin(client, user_auth_headers): """Test that non-admin users cannot perform group operations.""" # Try to create a group response = client.post( - "/api/v1/groups/", - json={"name": "test"}, - headers=user_auth_headers + "/api/v1/groups/", json={"name": "test"}, headers=user_auth_headers ) assert response.status_code == 403 diff --git a/test/test_services/test_user_manager.py b/test/test_services/test_user_manager.py index e1fef1f..f503acd 100644 --- a/test/test_services/test_user_manager.py +++ b/test/test_services/test_user_manager.py @@ -1,14 +1,10 @@ -import pytest -from fastapi import status - - def test_create_user(client, auth_headers): """Test creating a new user.""" user_data = { "name": "newuser", "email": "newuser@example.com", "is_admin": False, - "password": "newpassword123" + "password": "newpassword123", } response = client.post("/api/v1/users/", json=user_data, headers=auth_headers) @@ -27,7 +23,7 @@ def test_create_user_unauthorized(client): user_data = { "name": "unauthorized_user", "email": "unauthorized@example.com", - "password": "password123" + "password": "password123", } response = client.post("/api/v1/users/", json=user_data) @@ -56,6 +52,7 @@ def test_get_user_by_id(client, auth_headers, regular_user): assert data["id"] == regular_user.id assert data["name"] == regular_user.name + def test_get_current_user(client, auth_headers, admin_user): """Test getting the special url current""" response = client.get("/api/v1/users/current", headers=auth_headers) @@ -66,6 +63,7 @@ def test_get_current_user(client, auth_headers, admin_user): assert data["name"] == admin_user.name assert data["is_admin"] == admin_user.is_admin + def test_get_nonexistent_user(client, auth_headers): """Test retrieving a non-existent user.""" response = client.get("/api/v1/users/99999", headers=auth_headers) @@ -75,15 +73,10 @@ def test_get_nonexistent_user(client, auth_headers): def test_update_user(client, auth_headers, regular_user): """Test updating a user.""" - update_data = { - "name": "updated_name", - "email": "updated@example.com" - } + update_data = {"name": "updated_name", "email": "updated@example.com"} response = client.patch( - f"/api/v1/users/{regular_user.id}", - json=update_data, - headers=auth_headers + f"/api/v1/users/{regular_user.id}", json=update_data, headers=auth_headers ) assert response.status_code == 200 @@ -96,21 +89,17 @@ def test_update_user(client, auth_headers, regular_user): def test_update_user_password(client, auth_headers, regular_user): """Test updating a user's password.""" - update_data = { - "password": "new_password_456" - } + update_data = {"password": "new_password_456"} response = client.patch( - f"/api/v1/users/{regular_user.id}", - json=update_data, - headers=auth_headers + f"/api/v1/users/{regular_user.id}", json=update_data, headers=auth_headers ) assert response.status_code == 200 # Verify password can be used for login login_response = client.post( "/api/v1/token", - data={"username": regular_user.name, "password": "new_password_456"} + data={"username": regular_user.name, "password": "new_password_456"}, ) assert login_response.status_code == 200 @@ -118,7 +107,9 @@ def test_update_user_password(client, auth_headers, regular_user): def test_update_nonexistent_user(client, auth_headers): """Test updating a non-existent user.""" update_data = {"name": "updated"} - response = client.patch("/api/v1/users/99999", json=update_data, headers=auth_headers) + response = client.patch( + "/api/v1/users/99999", json=update_data, headers=auth_headers + ) assert response.status_code == 404 @@ -143,9 +134,9 @@ def test_user_operations_by_non_admin(client, user_auth_headers): """Test that non-admin users cannot perform admin operations.""" # Try to create a user response = client.post( - "/api/v1/users/", - json={"name": "test", "password": "pass"}, - headers=user_auth_headers + "/api/v1/users/", + json={"name": "test", "password": "pass"}, + headers=user_auth_headers, ) assert response.status_code == 403 -- 2.51.2 From fd8f2d164932ea0f5ca2227dd42359a3f5b7d827 Mon Sep 17 00:00:00 2001 From: ahtlon Date: Wed, 29 Jul 2026 02:41:38 +0200 Subject: [PATCH 9/9] This shouldn't ever have been comitted - oops --- test.py | 181 -------------------------------------------------------- 1 file changed, 181 deletions(-) delete mode 100644 test.py diff --git a/test.py b/test.py deleted file mode 100644 index 943726b..0000000 --- a/test.py +++ /dev/null @@ -1,181 +0,0 @@ -""" -This is a more involved example that performs initial configuration (often called personalization) of a DESFire card. - -It performs the following steps: -1. Authenticate with the default DES key -3. Change the default key -2. Create an application -4. Change the application master key -6. Create a read and write key (diversified) -7. Create an encrypted file -8. Write the UID to the encrypted file - -""" - -import logging -import os - -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 smartcard.CardRequest import CardRequest -from smartcard.CardType import AnyCardType -from smartcard.Exceptions import CardRequestTimeoutException - -# Please make sure to yet your own keys here before running this script -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") - -# Constants -MIFARE_APP_ID = "DEAFFE" # 7 bytes -MIFARE_ACL_READ_BASE_KEY_ID = 0x1 -MIFARE_ACL_WRITE_BASE_KEY_ID = 0x2 -MIFARE_SYS_ID = "FF0000" # 3 bytes, can essentially be anything -MIFARE_ENCRYPTED_FILE_ID = 0x1 - -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) - -cardtype = AnyCardType() -cardrequest = CardRequest(timeout=30, cardType=cardtype) -print("Please present DESfire tag...") - -try: - cardservice = cardrequest.waitforcard() -except CardRequestTimeoutException: - print("No tag detected within the timeout.") - raise - -cardservice.connection.connect() - -# Create Desfire object -desfire = DESFire(PCSCDevice(cardservice.connection.component)) - -# Create Key objects -AES_NULL_KEY_DATA = "00" * 16 -aes_keysettings = KeySettings( - key_type=DESFireKeyType.DF_KEY_AES, -) -aes_null_key = DESFireKey(aes_keysettings, AES_NULL_KEY_DATA) - -# Authenticate with default DES key -print("Authenticating with default DES key...") -key_settings = desfire.get_key_setting() -mk = DESFireKey(key_settings, "00" * 8) -desfire.authenticate(0x0, mk) - -# Get real UID -print("Getting real UID...") -uid = desfire.get_real_uid() -print(" - UID: ", to_hex_string(uid)) - -# Set default key -print("Setting default key...") -desfire.change_default_key(aes_null_key, 0x0) - -# Create application -print("Creating application...") -app_settings = KeySettings( - settings=[ - DESFireKeySettings.KS_ALLOW_CHANGE_MK, - DESFireKeySettings.KS_LISTING_WITHOUT_MK, - DESFireKeySettings.KS_CREATE_DELETE_WITHOUT_MK, - DESFireKeySettings.KS_CONFIGURATION_CHANGEABLE, - ], - key_type=DESFireKeyType.DF_KEY_AES, -) -desfire.create_application(MIFARE_APP_ID, app_settings, 4) - -# Verify application creation -applications = desfire.get_application_ids() -assert len(applications) == 1 -assert applications[0] == get_list(MIFARE_APP_ID) -print(" - Application created successfully.") - -# Select application -print("Selecting application...") -desfire.select_application(MIFARE_APP_ID) - -# Authenticate with AES key, as this has been set as the default key -print("Authenticating with AES key...") -# Create a new one as key data would be overriden by session data -aes_null_auth_key = DESFireKey(aes_keysettings, AES_NULL_KEY_DATA) -desfire.authenticate(0x0, aes_null_auth_key) - -# Change Application master key -print("Changing application master key (AMK)...") -aes_app_mk = DESFireKey(aes_keysettings, MIFARE_APP_MASTER_KEY) -desfire.change_key(0x0, aes_null_key, aes_app_mk, 0x1) - -# Re-Authenticate with new AES key -print("Re-authenticating with new AES key...") -desfire.authenticate(0x0, aes_app_mk) - -# Change file read and write keys (diversified) -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 -) - -print("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) - -print("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) - -print("Create encrypted file containing UID...") -file_settings = FileSettings( - file_size=8, - encryption=DESFireCommunicationMode.ENCRYPTED, - permissions=FilePermissions( - read_key=MIFARE_ACL_READ_BASE_KEY_ID, - write_key=MIFARE_ACL_WRITE_BASE_KEY_ID, - ), - file_type=DESFireFileType.MDFT_STANDARD_DATA_FILE, -) -desfire.create_standard_file(MIFARE_ENCRYPTED_FILE_ID, file_settings) - -print("Read and verify file settings again...") -file_data = desfire.get_file_settings(MIFARE_ENCRYPTED_FILE_ID) -assert file_data.file_size == 8 -assert file_data.encryption == DESFireCommunicationMode.ENCRYPTED -assert file_data.permissions is not None -assert file_data.permissions.read_access == MIFARE_ACL_READ_BASE_KEY_ID -assert file_data.permissions.write_access == MIFARE_ACL_WRITE_BASE_KEY_ID -assert file_data.file_type == DESFireFileType.MDFT_STANDARD_DATA_FILE -print(" - File created successfully.") - -print("Writing UID to encrypted file...") -data = [0x0] + uid -assert len(data) == 8 -desfire.write_file_data( - MIFARE_ENCRYPTED_FILE_ID, 0x0, file_data.encryption, get_list(data) -) - -print("Reading from encrypted file...") -rdata = desfire.read_file_data(MIFARE_ENCRYPTED_FILE_ID, file_data) -assert rdata == data -print(" - Data written successfully.") - -print("Personalization finished.") -- 2.51.2