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
16 changed files with 306 additions and 242 deletions

View File

@@ -1,39 +1,17 @@
## Gatekeeper - Door access system ## Gatekeeper - Door access system
#### Status: WIP - getting there o.o Status: WIP - not prod ready
Start prod server `nix run`<br> Dev with `nix develop`
Start dev server `nix run .#dev` or `nix run .#dev -- {args}`<br> sync python deps with `uv sync`
Interactive dev with `nix develop`, then sync deps with `uv sync`<br> start dev server `uv run fastapi dev`
Swagger UI @ http://127.0.0.1:8000/api/v1/docs<br> Swagger UI @ http://127.0.0.1:8000/docs
OpenApi @ http://127.0.0.1:8000/api/v1/openapi.json<br> start prod server `uv run fastapi run`
#### Config: You need to set services.pcscd.enable = true; for the smartcard reader to work
Nixos config for the card reader: Issues:
```
services.pcscd = {
enable = true;
plugins = [ pkgs.acsccid ]; #<-- Needed for the ACE1552U
};
```
This application is configured using env vars. You can use `cp .env.example .env` and then edit `.env` to change the keys to something else.
### A note on keycard scanners
I have tried two scanners while developing this project, the TWN4 MultiTech 2 from elatec and the ACR1552U from acs.<br>
##### TWN4
From the factory the TWN4 comes with firmware that emulates a HID and just inputs the serial of any scanned card. To be usable with pcscd it had to be flashed with the `TWN4_xPx520_S1SC161_Multi_CCID_1Slot_Standard.bix` firmware found in the `TWN4DevPack520` which can be downloaded under https://www.elatec-rfid.com/int/elatec-software (They send you a download link via email).
This software seems to not work under wine. I had to setup a win10 vm and pass the usb to be able to flash.<br>
Range: In my testing the TWN4 has a range of about 22mm when interfacing with our Mifare DESFire v2 cards. This is not enough for our idea of mounting the scanner on the inside of a glass window.
#### ACR1552U
The ACR1552U comes preloaded with PC/SC firmware. You need to install the acsccid package with your package-manager of choice (Nixos user see above) for the scanner to work but after that I had no problems with the hardware interfacing.<br>
Range: The range of the ACR1552U was much better at over 60mm (almost 70mm if you take of the face plate)
#### Issues:
- documentation missing - documentation missing
- `nix run` currently broken
- raspberry pi image not working - raspberry pi image not working
- no door state - no door state
- no door operations - no door operations

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

@@ -7,7 +7,7 @@ class Base(SQLModel):
#### 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
@@ -84,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": {

142
flake.nix
View File

@@ -21,15 +21,24 @@
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 =
{ {
self,
nixpkgs, nixpkgs,
pyproject-nix, pyproject-nix,
uv2nix, uv2nix,
pyproject-build-systems, pyproject-build-systems,
nixos-hardware,
... ...
}: }:
let let
@@ -107,22 +116,121 @@
default = pythonSets.${system}.mkVirtualEnv "gatekeeper" workspace.deps.default; default = pythonSets.${system}.mkVirtualEnv "gatekeeper" workspace.deps.default;
}); });
apps = forAllSystems (system: {
default = { nixosConfigurations = {
type = "app"; gatekeeper-rpi = nixpkgs.lib.nixosSystem {
program = toString (nixpkgs.legacyPackages.${system}.writeShellScript "gatekeeper" '' system = "aarch64-linux";
export LD_LIBRARY_PATH="${lib.getLib nixpkgs.legacyPackages.${system}.pcsclite}/lib''${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}" specialArgs = {
exec ${self.packages.${system}.default}/bin/fastapi run ${self}/app/main.py "$@"; 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";
})
];
}; };
dev = { };
type = "app";
program = toString (nixpkgs.legacyPackages.${system}.writeShellScript "gatekeeper" ''
export LD_LIBRARY_PATH="${lib.getLib nixpkgs.legacyPackages.${system}.pcsclite}/lib''${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}"
exec ${self.packages.${system}.default}/bin/fastapi dev ${self}/app/main.py "$@";
'');
};
});
nixosModules = { gatekeeper = import ./module.nix; };
}; };
} }

View File

@@ -1,41 +0,0 @@
{config, pkgs, lib, self, system, ... }:
let
cfg = config.services.gatekeeper;
in
{
options = {
services.gatekeeper = {
enable = lib.mkEnableOption "Enable the gatekeeper api service.";
dotenv = lib.mkOption {
type = lib.types.path;
default = null;
description = "The path to a .env file with the keys";
};
db = {
type = lib.types.path;
default = null;
description = "Where to save the database.";
};
};
};
config = lib.mkIf cfg.enable {
users.extraGroups.gatekeeper = {};
users.extraUsers.gatekeeper = {
description = "gatekeeper user";
group = "gatekeeper";
isSystemUser = true;
};
#environment.systemPackages = [self.packages.${system}.default];
systemd.services.gatekeeper = {
script = ''
export LD_LIBRARY_PATH="${lib.getLib pkgs.pcsclite}/lib''${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}"
exec ${pkgs.python3.pkgs.uvicorn}/bin/uvicorn run ${self}/app/main.py
'';
wantedBy = ["multi-user.target"];
serviceConfig = {
User = "gatekeeper";
Restart = "on-failure";
};
};
};
}

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)