14 Commits

Author SHA1 Message Date
1883662b5f Update README 2026-07-04 01:04:26 +02:00
a91bda16c2 [nix] nix run now possible 2026-07-04 01:04:15 +02:00
2d636f2bd7 [Main] add note about new doc url
Fixes #12
2026-07-01 17:54:16 +02:00
2531d4709b [Tests] add card manager PATCH tests 2026-07-01 17:47:19 +02:00
adbe64812a Merge pull request 'Card model changes' (#14) from change_card_model into master
Reviewed-on: #14

Fixes #5, #6, #13
2026-07-01 17:28:02 +02:00
098925c026 [Cards] Add patch endpoint 2026-07-01 17:25:18 +02:00
cd1c056c9a [Cards, Users] Check unique and return error 2026-06-27 19:17:16 +02:00
efa3cbfb58 [Tests] Fix various tests
add new fields to card tests; stop test_get_session from writing into prod db
2026-06-27 18:22:02 +02:00
b71e6388fa [Debug] fix debug router for new card model
(also remove_card_manually now actually removes cards...)
2026-06-27 17:12:40 +02:00
e2db1ecf37 [Cards] Add new fields to registration 2026-06-27 16:59:06 +02:00
7922e710cf [Cards] add serial, enabled, name fields 2026-06-27 16:39:23 +02:00
8268476053 [Users] avoid printing password to log 2026-06-27 16:38:13 +02:00
a32e820a02 [Card] change path operation to DELETE for #5 2026-06-27 15:55:10 +02:00
eec575c619 [Cards] Change all instances of uuid to key 2026-06-27 15:06:05 +02:00
14 changed files with 189 additions and 64 deletions

View File

@@ -1,17 +1,39 @@
## Gatekeeper - Door access system
Status: WIP - not prod ready
#### Status: WIP - getting there o.o
Dev with `nix develop`
sync python deps with `uv sync`
start dev server `uv run fastapi dev`
Swagger UI @ http://127.0.0.1:8000/docs
start prod server `uv run fastapi run`
Start prod server `nix run`<br>
Start dev server `nix run .#dev` or `nix run .#dev -- {args}`<br>
Interactive dev with `nix develop`, then sync deps with `uv sync`<br>
Swagger UI @ http://127.0.0.1:8000/api/v1/docs<br>
OpenApi @ http://127.0.0.1:8000/api/v1/openapi.json<br>
You need to set services.pcscd.enable = true; for the smartcard reader to work
#### Config:
Issues:
Nixos config for the card reader:
```
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
- `nix run` currently broken
- raspberry pi image not working
- no door state
- no door operations

View File

@@ -5,7 +5,7 @@ from sqlmodel import Session, select
from typing import List
from sqlalchemy.exc import NoResultFound
from ..model.models import Card
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
@@ -13,28 +13,40 @@ from app.services.scanner import WriteNewCard, DeleteCard
card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"])
def register_card(group_id: int):
key = WriteNewCard()
def register_card(cardInput: CardCreate):
key, uid = WriteNewCard()
if key == None:
logger.info("No card registered. Check logs!")
raise HTTPException(status.HTTP_417_EXPECTATION_FAILED, detail="No card registered. Check logs!")
card = Card(group_id=group_id, uuid=key)
card = Card(
group_id=cardInput.group_id,
key=key,
name=cardInput.name,
card_serial=uid,
enabled=cardInput.enabled
)
return card
@card_router.post("/{group_id}", response_model=Card)
def add_card(*, db: Session = Depends(get_session), group_id: int, admin: bool = Depends(auth_is_admin)):
card = register_card(group_id)
@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
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)
@card_router.get("/delete")
@card_router.delete("/")
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.uuid == key)).one()
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_code=500, 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"}
@@ -44,8 +56,15 @@ 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()
return cards
#TODO:
# -Split Authorisations + Cards
# -Deactivation
# -Deleting
@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)):
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
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,6 +3,7 @@ logger = logging.getLogger(__name__)
from fastapi import APIRouter, Depends, HTTPException
from app.services.auth import auth_is_admin
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
@@ -12,19 +13,31 @@ debug_router = APIRouter(
dependencies=[Depends(auth_is_admin)],
)
@debug_router.get("/addcard/{groupid}/{card_key}")
def add_card_manually(groupid: int, card_key: str, db: Session=Depends(get_session)):
@debug_router.put("/addcard/")
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}")
card = Card(group_id=groupid, uuid=card_key)
card = Card(
group_id=groupid,
key=card_key,
name=name,
enabled=enabled,
card_serial="00:00:00:00:00:00:00"
)
add_and_refresh(db, card)
return card
@debug_router.get("/rmcard/{card_key}")
def remove_card_manually(card_key: str, db: Session = Depends(get_session)):
logger.critical(f"Manual db change: removing a card with key: {card_key}")
card = db.exec(select(Card).where(Card.uuid == card_key)).one()
add_and_refresh(db, card)
@debug_router.get("/rmcard/{card_id}")
def remove_card_manually(card_id: str, db: Session = Depends(get_session)):
try:
card = db.exec(select(Card).where(Card.id == card_id)).one()
except exc.NoResultFound:
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")
def list_all_cards(db: Session = Depends(get_session)):
@@ -33,5 +46,5 @@ def list_all_cards(db: Session = Depends(get_session)):
print(cards)
out = []
for i in cards:
out.append({i.uuid: i.group.name})
out.append({i.key: i.group.name})
return out

View File

@@ -1,6 +1,6 @@
import logging
logger = logging.getLogger(__name__)
from fastapi import APIRouter, HTTPException, Depends
from fastapi import APIRouter, HTTPException, Depends, status
from sqlmodel import Session, select
from typing import List
@@ -12,8 +12,10 @@ 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)):
logger.info(f"creating user with data: {user}")
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)
return add_and_refresh(db, db_user)

View File

@@ -26,6 +26,9 @@ def checkDeps():
@asynccontextmanager
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()
create_db_and_tables()
create_first_user(db=get_db_session())

View File

@@ -7,7 +7,7 @@ class Base(SQLModel):
#### User
class UserBase(Base):
name: str = Field(index=True)
name: str = Field(index=True, unique=True)
email: str | None = None
is_admin: bool = False
@@ -84,10 +84,23 @@ class AccessAuthorizationUpdate(Base):
#### Card
class Card(Base, table=True):
id: int | None = Field(default=None, primary_key=True)
uuid: str
key: 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: 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

View File

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

View File

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

View File

@@ -25,6 +25,7 @@
outputs =
{
self,
nixpkgs,
pyproject-nix,
uv2nix,
@@ -105,5 +106,22 @@
packages = forAllSystems (system: {
default = pythonSets.${system}.mkVirtualEnv "gatekeeper" workspace.deps.default;
});
apps = forAllSystems (system: {
default = {
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 run ${self}/app/main.py "$@";
'');
};
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 run ${self}/app/main.py "$@";
'');
};
});
};
}

View File

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

View File

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

View File

@@ -23,9 +23,53 @@ 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/{test_group.id}", headers=user_auth_headers)
response = client.post(f"/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"
}
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,25 +26,16 @@ def test_create_db_and_tables():
def test_get_session(db_session):
"""Test database session generator."""
# Test that we can get a session
session_gen = get_session()
session = next(session_gen)
assert isinstance(session, Session)
assert isinstance(db_session, Session)
# Test that session works
user = UserDB(name="Test", passwordhash="hash")
session.add(user)
session.commit()
user = UserDB(name="Test_User", passwordhash="hash")
db_session.add(user)
db_session.commit()
retrieved_user = session.get(UserDB, user.id)
retrieved_user = db_session.get(UserDB, user.id)
assert retrieved_user is not None
assert retrieved_user.name == "Test"
# Clean up generator
try:
next(session_gen)
except StopIteration:
pass
assert retrieved_user.name == "Test_User"
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.commit()
card = Card(uuid="test-uuid-123", group_id=group.id)
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(
@@ -27,7 +27,7 @@ def test_check_access_with_valid_timetable(db_session):
db_session.commit()
# Test: access should be granted within time window
result = checkAccess("test-uuid-123", db_session)
result = checkAccess("test-key-123", db_session)
assert result == True
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.commit()
card = Card(uuid="test-uuid-123", group_id=group.id)
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(
@@ -52,10 +52,10 @@ def test_check_access_outside_hours(db_session):
group.accessauths = [aa]
db_session.commit()
result = checkAccess("test-uuid-123", db_session)
result = checkAccess("test-key-123", db_session)
assert result == False
def test_check_access_invalid_card(db_session):
# Should raise exception for non-existent card
with pytest.raises(Exception):
checkAccess("non-existent-uuid", db_session)
checkAccess("non-existent-key", db_session)