[nix] add module.nix

Fixes #12

README.md

@@ -1,17 +1,39 @@
 ## Gatekeeper - Door access system
-Status: WIP - not prod ready
+#### Status: WIP - getting there o.o
 
-Dev with `nix develop`   
+Start prod server `nix run`<br>
-sync python deps with `uv sync`   
+Start dev server `nix run .#dev` or  `nix run .#dev -- {args}`<br>
-start dev server `uv run fastapi dev`   
+Interactive dev with `nix develop`, then sync deps with `uv sync`<br>
-Swagger UI @ http://127.0.0.1:8000/docs  
+Swagger UI @ http://127.0.0.1:8000/api/v1/docs<br>
-start prod server `uv run fastapi run`
+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

app/controllers/cardManager.py

@@ -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):
+def register_card(cardInput: CardCreate):
-    key = WriteNewCard()
+    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)
+@card_router.post("/", response_model=Card)
-def add_card(*, db: Session = Depends(get_session), group_id: int, admin: bool = Depends(auth_is_admin)):
+def add_card(cardInput: CardCreate, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
-    card = register_card(group_id)
+    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
 
-
+@card_router.patch("/{card_id}", response_model=Card)
-#TODO:
+def update_card(card_id: int, cardInput: CardUpdate, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
-# -Split Authorisations + Cards
+    db_card = db.get(Card, card_id)
-# -Deactivation
+    if db_card is None:
-# -Deleting
+        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)

app/controllers/debugManager.py

@@ -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}")
+@debug_router.put("/addcard/")
-def add_card_manually(groupid: int, card_key: str, 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}")
-    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}")
+@debug_router.get("/rmcard/{card_id}")
-def remove_card_manually(card_key: str, db: Session = Depends(get_session)):
+def remove_card_manually(card_id: str, db: Session = Depends(get_session)):
-    logger.critical(f"Manual db change: removing a card with key: {card_key}")
+    try:
-    card = db.exec(select(Card).where(Card.uuid == card_key)).one()
+        card = db.exec(select(Card).where(Card.id == card_id)).one()
-    add_and_refresh(db, card)
+    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

app/controllers/userManager.py

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

app/main.py

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

app/model/models.py

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

app/services/door.py

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

app/services/scanner.py

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

flake.nix

@@ -25,6 +25,7 @@
 
   outputs =
     {
+      self,
       nixpkgs,
       pyproject-nix,
       uv2nix,
@@ -105,5 +106,23 @@
       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 dev ${self}/app/main.py "$@";
+            '');
+        };
+      });
+      nixosModules = { gatekeeper = import ./module.nix; };
     };
 }

module.nix

@@ -0,0 +1,41 @@
+{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";
+      };
+    };
+  };
+}

test/conftest.py

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

test/test_models.py

@@ -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)
+    card = Card(key="test-key", group_id=1)
-    assert card.uuid == "test-uuid"
+    assert card.key == "test-key"
     assert card.group_id == 1
 
 

test/test_services/test_card_manager.py

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

test/test_services/test_database.py

@@ -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()
+    assert isinstance(db_session, Session)
-    session = next(session_gen)
-
-    assert isinstance(session, Session)
 
     # Test that session works
-    user = UserDB(name="Test", passwordhash="hash")
+    user = UserDB(name="Test_User", passwordhash="hash")
-    session.add(user)
+    db_session.add(user)
-    session.commit()
+    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"
+    assert retrieved_user.name == "Test_User"
-
-    # Clean up generator
-    try:
-        next(session_gen)
-    except StopIteration:
-        pass
 
 
 def test_add_and_refresh(db_session):

test/test_services/test_door.py

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