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

just_found_out_about_linters #17

Merged
ahtlon merged 9 commits from just_found_out_about_linters into master 2026-07-29 02:46:50 +02:00
Conversation 0 Commits 9 Files Changed 28 +736 -564
28 changed files with 736 additions and 564 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.vscode/settings.json vendored
View File

@@ -1,4 +1,8 @@
{
"[python]": {
"editor.formatOnSave": true,
"editor.defaultFormatter": "charliermarsh.ruff"
},
"python.testing.pytestArgs": [
"test"
],

5
app/__init__.py
View File

@@ -1,6 +1,7 @@
from fastapi import FastAPI
from .controllers import userManager, cardManager
from .controllers import cardManager, userManager
app = FastAPI()
app.include_router(userManager.user_router)
app.include_router(cardManager.card_router)
app.include_router(cardManager.card_router)

106
app/controllers/aaManager.py
View File

@@ -1,51 +1,85 @@
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 app.model.models import *
from app.services.database import engine, get_session, add_and_refresh
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import selectinload
from sqlmodel import Session, select
from app.model.models import (
AccessAuthorizationCreate,
AccessAuthorizationDB,
AccessAuthorizationResponse,
AccessAuthorizationUpdate,
GroupDB,
GroupResponse,
OneShotAccess,
Timetable,
)
from app.services.auth import auth_is_admin
import uuid as gen_uuid
from app.services.database import add_and_refresh, get_session
logger = logging.getLogger(__name__)
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 is not []:
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)):
@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)
.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 +87,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 +113,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 +138,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"}
return {"message": "AccessAuthorization deleted successfully"}

90
app/controllers/cardManager.py
View File

@@ -1,70 +1,112 @@
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
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
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!")
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()
is 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)):
@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
@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)
return add_and_refresh(db, db_card)

37
app/controllers/debugManager.py
View File

@@ -1,11 +1,14 @@
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.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",
@@ -13,20 +16,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:
@@ -37,14 +50,14 @@ 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)):
logger.info(f"Debug Setting: Getting cards.")
logger.info("Debug Setting: Getting cards.")
cards = db.exec(select(Card)).all()
print(cards)
out = []
for i in cards:
out.append({i.key: i.group.name})
return out
return out

17
app/controllers/doorManager.py
View File

@@ -1,20 +1,25 @@
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"])
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)):
def close_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)
return doorService.checkAccess(input, db=db)

36
app/controllers/groupManager.py
View File

@@ -1,31 +1,47 @@
from fastapi import APIRouter, HTTPException, Depends, status
from fastapi import APIRouter, Depends, HTTPException, 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 ..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])
def get_groups(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
@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
@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"}
return {"message": "Group deleted successfully"}

68
app/controllers/userManager.py
View File

@@ -1,42 +1,73 @@
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
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select
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")
@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()
is 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)):
@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
@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")
@@ -48,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"}

37
app/main.py
View File

@@ -1,24 +1,34 @@
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())
logging.basicConfig(level=logging.INFO)
@asynccontextmanager
async def lifespan(app: FastAPI):
verify_settings()
@@ -28,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",
@@ -60,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)
app.include_router(debugManager.debug_router)

76
app/model/models.py
View File

@@ -1,62 +1,83 @@
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
#### 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)
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):
@@ -64,31 +85,43 @@ class AccessAuthorizationBase(Base):
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"] = []
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":
@@ -100,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"] = []
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
#### Card
class Card(Base, table=True):
@@ -124,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

66
app/services/auth.py
View File

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

13
app/services/database.py
View File

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

36
app/services/door.py
View File

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

144
app/services/scanner.py
View File

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

29
app/services/settings.py
View File

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

6
pyproject.toml
View File

@@ -17,6 +17,7 @@ dependencies = [
"setuptools>=82.0.1",
"pyscard>=2.3.1",
"alembic>=1.18.5",
"ruff>=0.16.0",
]
[tool.uv.sources]
@@ -28,3 +29,8 @@ python-desfire = ["poetry"]
[tool.setuptools]
py-modules = ["app"]
[tool.ruff]
exclude = [
"alembic"
]

164
test.py
View File

@@ -1,164 +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 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')
# 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.")

61
test/conftest.py
View File

@@ -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()

2
test/test_main.py
View File

@@ -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

28
test/test_models.py
View File

@@ -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

75
test/test_services/test_aa_manager.py
View File

@@ -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

37
test/test_services/test_auth.py
View File

@@ -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

33
test/test_services/test_card_manager.py
View File

@@ -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"]
assert "GroupID not found" in response.json()["detail"]

9
test/test_services/test_database.py
View File

@@ -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()

47
test/test_services/test_door.py
View File

@@ -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):

8
test/test_services/test_group_manager.py
View File

@@ -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

39
test/test_services/test_user_manager.py
View File

@@ -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

27
uv.lock generated
View File

@@ -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"