2 Commits

Author SHA1 Message Date
e871330041 Update flake.lock 2026-06-27 14:59:00 +02:00
2d64edde64 WIP add rpi nixos-config 2026-06-27 14:50:50 +02:00
14 changed files with 302 additions and 187 deletions

View File

@@ -5,7 +5,7 @@ from sqlmodel import Session, select
from typing import List from typing import List
from sqlalchemy.exc import NoResultFound from sqlalchemy.exc import NoResultFound
from ..model.models import Card, CardCreate, CardUpdate, GroupDB from ..model.models import Card
from ..services.database import engine, get_session, add_and_refresh from ..services.database import engine, get_session, add_and_refresh
from ..services.auth import auth_is_admin from ..services.auth import auth_is_admin
import uuid as gen_uuid import uuid as gen_uuid
@@ -13,40 +13,28 @@ from app.services.scanner import WriteNewCard, DeleteCard
card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"]) card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"])
def register_card(cardInput: CardCreate): def register_card(group_id: int):
key, uid = WriteNewCard() key = WriteNewCard()
if key == None: if key == None:
logger.info("No card registered. Check logs!") 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( card = Card(group_id=group_id, uuid=key)
group_id=cardInput.group_id,
key=key,
name=cardInput.name,
card_serial=uid,
enabled=cardInput.enabled
)
return card return card
@card_router.post("/", response_model=Card) @card_router.post("/{group_id}", response_model=Card)
def add_card(cardInput: CardCreate, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): def add_card(*, db: Session = Depends(get_session), group_id: int, admin: bool = Depends(auth_is_admin)):
try: assert db.exec(select(Card).where(Card.name == cardInput.name)).one_or_none() == None card = register_card(group_id)
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
except AssertionError:
raise HTTPException(status.HTTP_404_NOT_FOUND, detail="GroupID not found!")
card = register_card(cardInput)
return add_and_refresh(db, card) return add_and_refresh(db, card)
@card_router.delete("/") @card_router.get("/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() key = DeleteCard()
logger.info(key) logger.info(key)
try: try:
card = db.exec(select(Card).where(Card.key == key)).one() card = db.exec(select(Card).where(Card.uuid == key)).one()
except NoResultFound: except NoResultFound:
logger.info(f"The key:'{key}' was not found in db!") 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_code=500, detail="Key on card not found in DB. Please tell an admin about this. KEY={key}")
db.delete(card) db.delete(card)
db.commit() db.commit()
return {"message": "Card deleted successfully"} return {"message": "Card deleted successfully"}
@@ -56,15 +44,8 @@ def get_cards(*, db: Session = Depends(get_session), group_id: int, admin: bool
cards = db.exec(select(Card).where(Card.group_id == group_id)).all() cards = db.exec(select(Card).where(Card.group_id == group_id)).all()
return cards 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)): #TODO:
db_card = db.get(Card, card_id) # -Split Authorisations + Cards
if db_card is None: # -Deactivation
raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Card not found!") # -Deleting
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
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)

View File

@@ -3,7 +3,6 @@ logger = logging.getLogger(__name__)
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException
from app.services.auth import auth_is_admin from app.services.auth import auth_is_admin
from sqlmodel import Session, select from sqlmodel import Session, select
import sqlalchemy.exc as exc
from app.model.models import * from app.model.models import *
from app.services.database import get_session, add_and_refresh from app.services.database import get_session, add_and_refresh
@@ -13,31 +12,19 @@ debug_router = APIRouter(
dependencies=[Depends(auth_is_admin)], dependencies=[Depends(auth_is_admin)],
) )
@debug_router.put("/addcard/") @debug_router.get("/addcard/{groupid}/{card_key}")
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, db: Session=Depends(get_session)):
"""Add cards manually (you also have to delete them manually)""" """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( card = Card(group_id=groupid, uuid=card_key)
group_id=groupid,
key=card_key,
name=name,
enabled=enabled,
card_serial="00:00:00:00:00:00:00"
)
add_and_refresh(db, card) add_and_refresh(db, card)
return card return card
@debug_router.get("/rmcard/{card_id}") @debug_router.get("/rmcard/{card_key}")
def remove_card_manually(card_id: str, db: Session = Depends(get_session)): def remove_card_manually(card_key: str, db: Session = Depends(get_session)):
try: logger.critical(f"Manual db change: removing a card with key: {card_key}")
card = db.exec(select(Card).where(Card.id == card_id)).one() card = db.exec(select(Card).where(Card.uuid == card_key)).one()
except exc.NoResultFound: add_and_refresh(db, card)
raise HTTPException(status_code=500, detail="No card with that id found.")
logger.critical(f"Manual db change: removing a card with attrs: {card}")
db.delete(card)
db.commit()
return {"message": "Card deleted successfully"}
@debug_router.put("/getcards") @debug_router.put("/getcards")
def list_all_cards(db: Session = Depends(get_session)): def list_all_cards(db: Session = Depends(get_session)):
@@ -46,5 +33,5 @@ def list_all_cards(db: Session = Depends(get_session)):
print(cards) print(cards)
out = [] out = []
for i in cards: for i in cards:
out.append({i.key: i.group.name}) out.append({i.uuid: i.group.name})
return out return out

View File

@@ -1,6 +1,6 @@
import logging import logging
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
from fastapi import APIRouter, HTTPException, Depends, status from fastapi import APIRouter, HTTPException, Depends
from sqlmodel import Session, select from sqlmodel import Session, select
from typing import List from typing import List
@@ -12,10 +12,8 @@ user_router = APIRouter(tags=["Users"], prefix="/api/v1/users")
@user_router.post("/", response_model=UserResponse) @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)):
logger.info(f"creating user with data: {user}")
hashed_password = {"passwordhash": get_password_hash(user.password)} hashed_password = {"passwordhash": get_password_hash(user.password)}
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) db_user = UserDB.model_validate(user, update=hashed_password)
return add_and_refresh(db, db_user) return add_and_refresh(db, db_user)

View File

@@ -26,9 +26,6 @@ def checkDeps():
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI): async def lifespan(app: FastAPI):
logger.critical("-"*63)
logger.critical("---- Documentation is at http://127.0.0.1:8000/api/v1/docs ----")
logger.critical("-"*63)
checkDeps() checkDeps()
create_db_and_tables() create_db_and_tables()
create_first_user(db=get_db_session()) create_first_user(db=get_db_session())

View File

@@ -1,14 +1,13 @@
from sqlmodel import Field, Relationship, Session, SQLModel from sqlmodel import Field, Relationship, Session, SQLModel
from typing import List, Literal, Union from typing import List
from datetime import datetime, time from datetime import time
from pydantic import model_validator
class Base(SQLModel): class Base(SQLModel):
pass pass
#### User #### User
class UserBase(Base): class UserBase(Base):
name: str = Field(index=True, unique=True) name: str = Field(index=True)
email: str | None = None email: str | None = None
is_admin: bool = False is_admin: bool = False
@@ -61,49 +60,19 @@ class GroupResponse(GroupBase):
#### AccessAuthorization #### AccessAuthorization
class AccessAuthorizationBase(Base): class AccessAuthorizationBase(Base):
name: str = Field(index=True) name: str = Field(index=True)
type: Literal["timetable", "oneshot", "somefuturespec"]
is_active: bool is_active: bool
class AccessAuthorizationDB(AccessAuthorizationBase, table=True): class AccessAuthorizationDB(AccessAuthorizationBase, table=True):
id: int | None = Field(default=None, primary_key=True) id: int | None = Field(default=None, primary_key=True)
type: str
groups: List["GroupDB"] = Relationship(back_populates="accessauths", link_model=AaGroupLink) groups: List["GroupDB"] = Relationship(back_populates="accessauths", link_model=AaGroupLink)
timetables: List["Timetable"] = Relationship(back_populates="accessauth", cascade_delete=True) 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): 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")
if self.oneshot is not None:
raise ValueError("timetable auths are not allowed oneshot objects")
elif self.type == "oneshot":
if not self.oneshot:
raise ValueError("oneshot auths require a oneshot object")
if self.timetables:
raise ValueError("oneshot auths are not allowed timetable objects")
elif self.type == "somefuturespec":
raise ValueError("somefuturespec is not jet implemented. Please do not use")
return self
class AccessAuthorizationResponse(AccessAuthorizationBase): class AccessAuthorizationResponse(AccessAuthorizationBase):
id: int id: int
timetables: List["Timetable"] = [] timetables: List["Timetable"]
oneshot: OneShotAccessBase | None = None
groups: List["GroupDB"] groups: List["GroupDB"]
class AccessAuthorizationUpdate(Base): class AccessAuthorizationUpdate(Base):
@@ -115,23 +84,10 @@ class AccessAuthorizationUpdate(Base):
#### Card #### Card
class Card(Base, table=True): class Card(Base, table=True):
id: int | None = Field(default=None, primary_key=True) id: int | None = Field(default=None, primary_key=True)
key: str uuid: str
card_serial: str
enabled: bool = True
name: str = Field(unique=True, max_length=32)
group_id: int | None = Field(default=None, foreign_key="groupdb.id") group_id: int | None = Field(default=None, foreign_key="groupdb.id")
group: GroupDB | None = Relationship(back_populates="cards") 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): class TimetableBase(Base):
weekday: int = Field(le=6, ge=0) weekday: int = Field(le=6, ge=0)
starttime: time starttime: time

View File

@@ -28,11 +28,11 @@ def closeDoor():
def isDoorOpen(): def isDoorOpen():
return doorIsOpen return doorIsOpen
def checkAccess(key: str, db: Session): def checkAccess(uuid: str, db: Session):
try: try:
current_weekday = datetime.datetime.weekday(datetime.date.today()) current_weekday = datetime.datetime.weekday(datetime.date.today())
current_time = datetime.datetime.now() current_time = datetime.datetime.now()
card = db.exec(select(Card).where(Card.key == key)).one() card = db.exec(select(Card).where(Card.uuid == uuid)).one()
for auth in card.group.accessauths: for auth in card.group.accessauths:
logger.info(f"checking auth: {auth.name}") logger.info(f"checking auth: {auth.name}")
for timetable in auth.timetables: for timetable in auth.timetables:

View File

@@ -233,7 +233,7 @@ def WriteNewCard():
assert rdata == get_list(key) assert rdata == get_list(key)
logger.debug(" - Data written successfully.") logger.debug(" - Data written successfully.")
scannerThread.start() scannerThread.start()
return key, to_hex_string(data=uid, separator=":") return key
except Exception as e: except Exception as e:
logger.error(f"Error in write function: {e}", exc_info=True) logger.error(f"Error in write function: {e}", exc_info=True)

128
flake.lock generated
View File

@@ -1,12 +1,114 @@
{ {
"nodes": { "nodes": {
"argononed": {
"flake": false,
"locked": {
"lastModified": 1729566243,
"narHash": "sha256-DPNI0Dpk5aym3Baf5UbEe5GENDrSmmXVdriRSWE+rgk=",
"owner": "nvmd",
"repo": "argononed",
"rev": "16dbee54d49b66d5654d228d1061246b440ef7cf",
"type": "github"
},
"original": {
"owner": "nvmd",
"repo": "argononed",
"type": "github"
}
},
"flake-compat": {
"locked": {
"lastModified": 1767039857,
"narHash": "sha256-vNpUSpF5Nuw8xvDLj2KCwwksIbjua2LZCqhV1LNRDns=",
"owner": "edolstra",
"repo": "flake-compat",
"rev": "5edf11c44bc78a0d334f6334cdaf7d60d732daab",
"type": "github"
},
"original": {
"owner": "edolstra",
"repo": "flake-compat",
"type": "github"
}
},
"nixos-hardware": {
"inputs": {
"nixpkgs": [
"nixpkgs"
]
},
"locked": {
"lastModified": 1782562157,
"narHash": "sha256-a7+T6QSeowynwZ1ZJJbP8T8ntAytvrui8kFGJmIZt2c=",
"owner": "NixOS",
"repo": "nixos-hardware",
"rev": "a9cf7546a938c737b079e738de73934a13de9784",
"type": "github"
},
"original": {
"owner": "NixOS",
"ref": "master",
"repo": "nixos-hardware",
"type": "github"
}
},
"nixos-images": {
"inputs": {
"nixos-stable": [
"nixos-raspberrypi",
"nixpkgs"
],
"nixos-unstable": [
"nixos-raspberrypi",
"nixpkgs"
]
},
"locked": {
"lastModified": 1747747741,
"narHash": "sha256-LUOH27unNWbGTvZFitHonraNx0JF/55h30r9WxqrznM=",
"owner": "nvmd",
"repo": "nixos-images",
"rev": "cbbd6db325775096680b65e2a32fb6187c09bbb4",
"type": "github"
},
"original": {
"owner": "nvmd",
"ref": "sdimage-installer",
"repo": "nixos-images",
"type": "github"
}
},
"nixos-raspberrypi": {
"inputs": {
"argononed": "argononed",
"flake-compat": "flake-compat",
"nixos-images": "nixos-images",
"nixpkgs": [
"nixpkgs"
]
},
"locked": {
"lastModified": 1779023229,
"narHash": "sha256-MInilg7B/06c34SwOuGSBho4l0H1EZcmvxTkSWCs5pE=",
"owner": "nvmd",
"repo": "nixos-raspberrypi",
"rev": "06c6e3513e1ee64b651913193fc6ac38aa4963f5",
"type": "github"
},
"original": {
"owner": "nvmd",
"ref": "main",
"repo": "nixos-raspberrypi",
"type": "github"
}
},
"nixpkgs": { "nixpkgs": {
"locked": { "locked": {
"lastModified": 1778869304, "lastModified": 1782467914,
"narHash": "sha256-30sZNZoA1cqF5JNO9fVX+wgiQYjB7HJqqJ4ztCDeBZE=", "narHash": "sha256-pGvFkM8N0xEkIIXDe5YYfbEAvHrk4IxBrjB/x8OomhE=",
"owner": "nixos", "owner": "nixos",
"repo": "nixpkgs", "repo": "nixpkgs",
"rev": "d233902339c02a9c334e7e593de68855ad26c4cb", "rev": "e73de5be04e0eff4190a1432b946d469c794e7b4",
"type": "github" "type": "github"
}, },
"original": { "original": {
@@ -29,11 +131,11 @@
] ]
}, },
"locked": { "locked": {
"lastModified": 1776659114, "lastModified": 1782093830,
"narHash": "sha256-qapCOQmR++yZSY43dzrp3wCrkOTLpod+ONtJWBk6iKU=", "narHash": "sha256-6gmEVe69+KlRkZD4PEEV5xAlB9CB0Y9TiuEgQjDrKTQ=",
"owner": "pyproject-nix", "owner": "pyproject-nix",
"repo": "build-system-pkgs", "repo": "build-system-pkgs",
"rev": "ffaa2161dd5d63e0e94591f86b54fc239660fb2e", "rev": "430680a19bc85a3bda55f12e4cc1a1aadcf2e478",
"type": "github" "type": "github"
}, },
"original": { "original": {
@@ -49,11 +151,11 @@
] ]
}, },
"locked": { "locked": {
"lastModified": 1778901413, "lastModified": 1782089418,
"narHash": "sha256-GSKXTAnFqRAMlZkJrIPcQMYf+lpMr66K3i60mB9STvc=", "narHash": "sha256-LRD1SuQWr49fGq3A+8GLXfsLE2xqIpQA440YDZwms3M=",
"owner": "pyproject-nix", "owner": "pyproject-nix",
"repo": "pyproject.nix", "repo": "pyproject.nix",
"rev": "a228447c3e179d477c1b6246ef3efa8cfe3c469a", "rev": "43f0b40edd0a74c63f66b7b48d969ae6b740d611",
"type": "github" "type": "github"
}, },
"original": { "original": {
@@ -64,6 +166,8 @@
}, },
"root": { "root": {
"inputs": { "inputs": {
"nixos-hardware": "nixos-hardware",
"nixos-raspberrypi": "nixos-raspberrypi",
"nixpkgs": "nixpkgs", "nixpkgs": "nixpkgs",
"pyproject-build-systems": "pyproject-build-systems", "pyproject-build-systems": "pyproject-build-systems",
"pyproject-nix": "pyproject-nix", "pyproject-nix": "pyproject-nix",
@@ -80,11 +184,11 @@
] ]
}, },
"locked": { "locked": {
"lastModified": 1779269674, "lastModified": 1782100052,
"narHash": "sha256-P1LHCRdYpdtHAEzuEsNHrI6d9mVPl5a2fyFDZGHNVbI=", "narHash": "sha256-UfyLY3Hfwb3JqxCcj0953GxTFI5dEL85EKEe6DAFiVs=",
"owner": "pyproject-nix", "owner": "pyproject-nix",
"repo": "uv2nix", "repo": "uv2nix",
"rev": "69aec536f6d1acc415ed2e20299312802aba98c6", "rev": "920fc6dfaf9f10ec56de93b184055e1a9f380d5e",
"type": "github" "type": "github"
}, },
"original": { "original": {

127
flake.nix
View File

@@ -21,6 +21,15 @@
inputs.uv2nix.follows = "uv2nix"; inputs.uv2nix.follows = "uv2nix";
inputs.nixpkgs.follows = "nixpkgs"; inputs.nixpkgs.follows = "nixpkgs";
}; };
nixos-hardware = {
url = "github:NixOS/nixos-hardware/master";
inputs.nixpkgs.follows = "nixpkgs";
};
nixos-raspberrypi = {
url = "github:nvmd/nixos-raspberrypi/main";
inputs.nixpkgs.follows = "nixpkgs";
};
}; };
outputs = outputs =
@@ -29,6 +38,7 @@
pyproject-nix, pyproject-nix,
uv2nix, uv2nix,
pyproject-build-systems, pyproject-build-systems,
nixos-hardware,
... ...
}: }:
let let
@@ -105,5 +115,122 @@
packages = forAllSystems (system: { packages = forAllSystems (system: {
default = pythonSets.${system}.mkVirtualEnv "gatekeeper" workspace.deps.default; default = pythonSets.${system}.mkVirtualEnv "gatekeeper" workspace.deps.default;
}); });
nixosConfigurations = {
gatekeeper-rpi = nixpkgs.lib.nixosSystem {
system = "aarch64-linux";
specialArgs = {
inherit workspace overlay;
};
modules = [
"${nixpkgs}/nixos/modules/profiles/headless.nix"
"${nixpkgs}/nixos/modules/profiles/minimal.nix"
"${nixpkgs}/nixos/modules/installer/sd-card/sd-image-aarch64.nix"
nixos-hardware.nixosModules.raspberry-pi-3
({ pkgs, ... }: {
nix.nixPath = [
"nixpkgs=${pkgs.path}"
];
nix.settings = {
trusted-users = [ "root" "@wheel" ];
experimental-features = [ "nix-command" "flakes" ];
auto-optimise-store = true;
};
})
({ config, pkgs, packages, ...}:
let
gatekeeper-app = pkgs.python3Packages.buildPythonPackage {
pname = "gatekeeper-app";
version = "0.0.1";
propagatedBuildInputs = [packages.gatekeeper];
src = ./app;
format = "pyproject";
};
in
{
boot.loader.grub.enable = false;
boot.loader.generic-extlinux-compatible.enable = true;
boot.kernelPackages = pkgs.linuxPackages;
boot.tmp.cleanOnBoot = true;
hardware.enableRedistributableFirmware = true;
networking.hostName = "gatekeeper-rpi";
#minimal
documentation.man.enable = false;
xdg.icons.enable = false;
xdg.mime.enable = false;
xdg.sounds.enable = false;
users.users.root.openssh.authorizedKeys.keys = ["ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAICKaEcGaSKU0xC5qCwzj2oCLLG4PYjWHZ7/CXHw4urVk atlan@nixos"];
users.groups = { gatekeeper = {}; };
users.users.admin = {
isNormalUser = true;
extraGroups = [ "wheel" "networkmanager" "gatekeeper" ];
};
users.users.gatekeeper = {
isNormalUser = true;
extraGroups = ["gatekeeper"];
};
services.openssh = {
enable = true;
ports = [ 22 ];
settings = {
PasswordAuthentication = false;
};
};
environment.systemPackages = with pkgs; [
git
htop
pcsclite
gatekeeper-app
];
environment.etc."gatekeeper-env".source =
pythonSets.aarch64-linux.mkVirtualEnv "gatekeeper" workspace.deps.default;
services.pcscd.enable = true;
systemd.services.gatekeeper = {
enable = true;
description = "Gatekeeper service";
after = [ "network.target" "pcscd.service"];
wantedBy = [ "multi-user.target" ];
environment = {
LD_LIBRARY_PATH = "${lib.getLib pkgs.pcsclite}/lib";
PYTHONPATH = "/etc/gatekeeper-env/lib/python3.x/site-packages";
};
serviceConfig = {
Type = "simple";
User = "gatekeeper";
Group = "gatekeeper";
WorkingDirectory = "/var/lib/gatekeeper";
ExecStart = "${pythonSets.aarch64-linux.mkVirtualEnv "gatekeeper" workspace.deps.default}/bin/python -m uvicorn main:app --host 0.0.0.0 --port 8000";
Restart = "always";
RestartSec = "10";
};
# Create working directory and log directory
preStart = ''
mkdir -p /var/lib/gatekeeper
mkdir -p /var/log/gatekeeper
chown -R gatekeeper:gatekeeper /var/lib/gatekeeper
chown -R gatekeeper:gatekeeper /var/log/gatekeeper
'';
};
networking.firewall.allowedTCPPorts = [ 8000 ];
time.timeZone = "Europe/Berlin";
system.stateVersion = "25.11";
})
];
};
};
}; };
} }

View File

@@ -99,7 +99,7 @@ def test_group(db_session):
@pytest.fixture @pytest.fixture
def test_card(db_session, test_group): def test_card(db_session, test_group):
"""Create a test card.""" """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(uuid="test-uuid-123", group_id=test_group.id)
db_session.add(card) db_session.add(card)
db_session.commit() db_session.commit()
db_session.refresh(card) db_session.refresh(card)

View File

@@ -59,8 +59,8 @@ def test_access_authorization_models():
def test_card_model(): def test_card_model():
"""Test card model creation and validation.""" """Test card model creation and validation."""
card = Card(key="test-key", group_id=1) card = Card(uuid="test-uuid", group_id=1)
assert card.key == "test-key" assert card.uuid == "test-uuid"
assert card.group_id == 1 assert card.group_id == 1

View File

@@ -23,53 +23,9 @@ def test_get_cards_for_nonexistent_group(client, auth_headers):
def test_card_operations_by_non_admin(client, test_group, user_auth_headers): def test_card_operations_by_non_admin(client, test_group, user_auth_headers):
"""Test that non-admin users cannot perform card operations.""" """Test that non-admin users cannot perform card operations."""
# Try to add a card # Try to add a card
response = client.post(f"/api/v1/cards/", headers=user_auth_headers) response = client.post(f"/api/v1/cards/{test_group.id}", headers=user_auth_headers)
assert response.status_code == 403 assert response.status_code == 403
# Try to get cards # Try to get cards
response = client.get(f"/api/v1/cards/{test_group.id}", headers=user_auth_headers) response = client.get(f"/api/v1/cards/{test_group.id}", headers=user_auth_headers)
assert response.status_code == 403 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"
}
response = client.patch(
f"/api/v1/cards/{test_card.id}",
json=update_data,
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert data["name"] == "changed_name"
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
)
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"
}
response = client.patch(
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"]

View File

@@ -26,16 +26,25 @@ def test_create_db_and_tables():
def test_get_session(db_session): def test_get_session(db_session):
"""Test database session generator.""" """Test database session generator."""
# Test that we can get a session # Test that we can get a session
assert isinstance(db_session, Session) session_gen = get_session()
session = next(session_gen)
assert isinstance(session, Session)
# Test that session works # Test that session works
user = UserDB(name="Test_User", passwordhash="hash") user = UserDB(name="Test", passwordhash="hash")
db_session.add(user) session.add(user)
db_session.commit() session.commit()
retrieved_user = db_session.get(UserDB, user.id) retrieved_user = session.get(UserDB, user.id)
assert retrieved_user is not None assert retrieved_user is not None
assert retrieved_user.name == "Test_User" assert retrieved_user.name == "Test"
# Clean up generator
try:
next(session_gen)
except StopIteration:
pass
def test_add_and_refresh(db_session): def test_add_and_refresh(db_session):

View File

@@ -9,7 +9,7 @@ def test_check_access_with_valid_timetable(db_session):
db_session.add(group) db_session.add(group)
db_session.commit() 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(uuid="test-uuid-123", group_id=group.id)
db_session.add(card) db_session.add(card)
timetable = Timetable( timetable = Timetable(
@@ -27,7 +27,7 @@ def test_check_access_with_valid_timetable(db_session):
db_session.commit() db_session.commit()
# Test: access should be granted within time window # Test: access should be granted within time window
result = checkAccess("test-key-123", db_session) result = checkAccess("test-uuid-123", db_session)
assert result == True assert result == True
def test_check_access_outside_hours(db_session): def test_check_access_outside_hours(db_session):
@@ -36,7 +36,7 @@ def test_check_access_outside_hours(db_session):
db_session.add(group) db_session.add(group)
db_session.commit() 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(uuid="test-uuid-123", group_id=group.id)
db_session.add(card) db_session.add(card)
timetable = Timetable( timetable = Timetable(
@@ -52,10 +52,10 @@ def test_check_access_outside_hours(db_session):
group.accessauths = [aa] group.accessauths = [aa]
db_session.commit() db_session.commit()
result = checkAccess("test-key-123", db_session) result = checkAccess("test-uuid-123", db_session)
assert result == False assert result == False
def test_check_access_invalid_card(db_session): def test_check_access_invalid_card(db_session):
# Should raise exception for non-existent card # Should raise exception for non-existent card
with pytest.raises(Exception): with pytest.raises(Exception):
checkAccess("non-existent-key", db_session) checkAccess("non-existent-uuid", db_session)