README.md

@@ -1,9 +1,10 @@
 ## Gatekeeper - Door access system
-#### Status: WIP - getting there o.o
+#### Status: "WORKING" - Base functionality is there, mayor issues
 
 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>
+There is a nix module you can use by importing `inputs.gatekeeper.nixosModules.gatekeeper`<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>
 
@@ -33,17 +34,13 @@ Range: The range of the ACR1552U was much better at over 60mm (almost 70mm if yo
 
 
 #### Issues:
+   - cards can only unlock, not lock
    - documentation missing
-   - raspberry pi image not working
-   - no door state
-   - no door operations
-   - hardcoded secret key in auth.py -> centralise env var loading 
    - i don't like the error handling in the scanner - doesn't pass errors correctly
    - cors for frontend: https://fastapi.tiangolo.com/tutorial/cors
      - Load cors from env var or something
    - BackgroundScanner shouldn't get a single session for the whole lifecycle
    - input validation maybe
-   - too many imports
    - inconsistent logging (request logging?)
    - rate limiting maybe
    - pretty sure the controllers are doing too much stuff

app/__init__.py

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

app/controllers/doorManager.py

@@ -10,14 +10,21 @@ door_router = APIRouter(prefix="/api/v1/door", tags=["Door"])
 
 @door_router.put("/open")
 def open_door(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
-    doorService.opendoor()
+    doorService.openDoor()
 
 
 @door_router.put("/close")
 def close_door(
     db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
 ):
-    doorService.closedoor()
+    doorService.closeDoor()
+
+
+@door_router.put("/status")
+def is_door_open(
+    db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
+):
+    return doorService.isDoorOpen()
 
 
 @door_router.post("/test")

app/main.py

@@ -1,15 +1,10 @@
 import logging
-import os
 from contextlib import asynccontextmanager
 
-from dotenv import load_dotenv
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.security import OAuth2PasswordBearer
 
-load_dotenv()
-
-# ruff: disable[E402]
 from app.controllers import (
     aaManager,
     cardManager,
@@ -20,10 +15,9 @@ from app.controllers import (
 )
 from app.services.auth import create_first_user, token_router
 from app.services.database import create_db_and_tables, get_db_session
+from app.services.door import DoorController, init_controller
 from app.services.scanner import BackgroundScanner
-from app.services.settings import verify_settings
+from app.services.settings import settings
-
-# ruff: enable[E402]
 
 logger = logging.getLogger(__name__)
 
@@ -34,11 +28,16 @@ logging.basicConfig(level=logging.INFO)
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    verify_settings()
     create_db_and_tables()
     create_first_user(db=get_db_session())
-
+    init_controller(
-    if not os.getenv("DISABLE_CARDS"):
+        DoorController(
+            lock_pin=settings.lock_pin,
+            unlock_pin=settings.unlock_pin,
+            mock_factory=settings.mock_gpio,
+        ),
+    )
+    if not settings.disable_cards:
         scanner.start()
 
     logger.info("-" * 63)

app/services/auth.py

@@ -1,5 +1,4 @@
 import logging
-import os
 import secrets
 import string
 from datetime import UTC, datetime, timedelta
@@ -14,10 +13,11 @@ from sqlmodel import Session, select
 
 from app.model.models import Token, TokenData, UserDB
 from app.services.database import add_and_refresh, get_session
+from app.services.settings import settings
 
 logger = logging.getLogger(__name__)
 
-SECRET_KEY = os.getenv("SECRET_KEY", default="ff" * 16)
+SECRET_KEY = settings.secret_key
 ALGORITHM = "HS256"
 ACCESS_TOKEN_EXPIRE_MINUTES = 120
 

app/services/database.py

@@ -1,38 +1,48 @@
 import logging
-from os import getenv, path
+from functools import lru_cache
 
+from sqlalchemy import inspect
 from sqlmodel import Session, SQLModel, create_engine
 
+from app.services.settings import settings
+
 logger = logging.getLogger(__name__)
 
-SQLALCHEMY_DATABASE_PATH = getenv("SQLALCHEMY_DATABASE_PATH", "./gatekeeper.db")
-SQLALCHEMY_DATABASE_URL = "sqlite:///" + SQLALCHEMY_DATABASE_PATH
 
-engine = create_engine(SQLALCHEMY_DATABASE_URL)
+@lru_cache
+def get_engine():
+    return create_engine(
+        settings.sqlalchemy_database_url, connect_args={"check_same_thread": False}
+    )
 
 
 def create_db_and_tables():
-    if not path.exists(SQLALCHEMY_DATABASE_PATH):
+    inspector = inspect(get_engine())
-        SQLModel.metadata.create_all(engine)
+    existing_tables = inspector.get_table_names()
+    if not existing_tables:
+        SQLModel.metadata.create_all(get_engine())
         from alembic.config import Config
 
         from alembic import command
 
-        alembic_cfg = Config("./alembic.ini")
+        alembic_cfg = Config(settings.alembic_config)
-        alembic_cfg.attributes["sqlalchemy.url"] = SQLALCHEMY_DATABASE_URL
+        alembic_cfg.set_main_option("sqlalchemy.url", str(get_engine().url))
+        alembic_cfg.attributes["sqlalchemy.url"] = settings.sqlalchemy_database_url
         command.stamp(alembic_cfg, "head")
         logger.info("Database created and tables initialized.")
     else:
-        logger.info("Database already exists")
+        logger.info(
+            "Database already initialized (%d tables found).", len(existing_tables)
+        )
 
 
 def get_session():
-    with Session(engine) as db:
+    with Session(get_engine()) as db:
         yield db
 
 
 def get_db_session():
-    return Session(engine)
+    return Session(get_engine())
 
 
 def add_and_refresh(db: Session, obj):

app/services/door.py

@@ -1,6 +1,8 @@
 import logging
 from datetime import date, datetime, timedelta
+from time import sleep
 
+import lgpio
 from sqlalchemy import exc
 from sqlmodel import select
 
@@ -9,25 +11,81 @@ from app.services.database import Session, add_and_refresh
 
 logger = logging.getLogger(__name__)
 
-doorIsOpen = True
+
-# I think this could also be gpio controlled
 # See: https://github.com/technyon/nuki_hub#gpio-lock-control-optional
+# TODO: add sensor pin
+class DoorController:
+    def __init__(
+        self,
+        lock_pin: int = 17,  # connected to 20 on the esp
+        unlock_pin: int = 18,  # connected to 21 on the esp
+        mock_factory: bool = False,
+    ):
+        self._is_open: bool = False
+        self._lock_pin = lock_pin
+        self._unlock_pin = unlock_pin
+        self._mock = mock_factory
+        self._chip = None
+
+        if not mock_factory:
+            self._chip = lgpio.gpiochip_open(0)
+            lgpio.gpio_claim_output(self._chip, unlock_pin, 1)
+            lgpio.gpio_claim_output(self._chip, lock_pin, 1)
+
+        logger.info(
+            "DoorController started. lock=%s unlock=%s mock=%s",
+            lock_pin,
+            unlock_pin,
+            mock_factory,
+        )
+
+    def open(self):
+        if self._mock:
+            self._is_open = True
+            logger.info("Dor unlocked.[MOCK]")
+            return
+
+        lgpio.gpio_write(self._chip, self._unlock_pin, 0)
+        sleep(0.4)
+        lgpio.gpio_write(self._chip, self._unlock_pin, 1)
+        self._is_open = True
+        logger.info("Door unlocked!")
+
+    def close(self):
+        lgpio.gpio_write(self._chip, self._lock_pin, 0)
+        sleep(0.4)
+        lgpio.gpio_write(self._chip, self._lock_pin, 1)
+        self._is_open = False
+        logger.info("Door locked!")
+
+    def is_open(self):
+        return self._is_open
+
+
+_contoller: DoorController | None = None
+
+
+def init_controller(ctrl: DoorController):
+    global _contoller
+    _contoller = ctrl
+
+
+def get_controller():
+    if _contoller is None:
+        raise RuntimeError("DoorController not initialized.")
+    return _contoller
 
 
 def openDoor():
-    global doorIsOpen
+    get_controller().open()
-    doorIsOpen = True
-    logger.info("Still needs gpio out")
 
 
 def closeDoor():
-    global doorIsOpen
+    get_controller().close()
-    doorIsOpen = False
-    logger.info("Still needs gpio out")
 
 
 def isDoorOpen():
-    return doorIsOpen
+    return get_controller().is_open()
 
 
 def decrementOneshot(db: Session, oneshot: OneShotAccess):

app/services/scanner.py

@@ -1,5 +1,4 @@
 import logging
-import os
 import secrets
 import threading
 import time
@@ -25,15 +24,16 @@ from smartcard.CardRequest import CardRequest
 from smartcard.CardType import AnyCardType
 from smartcard.Exceptions import CardRequestTimeoutException
 
-from app.services.door import checkAccess, openDoor
+from app.services.door import checkAccess, closeDoor, isDoorOpen, openDoor
+from app.services.settings import settings
 
 logger = logging.getLogger(__name__)
 
 # ENV vars
 load_dotenv()
-MIFARE_APP_MASTER_KEY = os.getenv("MIFARE_APP_MASTER_KEY")
+MIFARE_APP_MASTER_KEY = settings.mifare_app_master_key
-MIFARE_ACL_READ_BASE_KEY = os.getenv("MIFARE_ACL_READ_BASE_KEY")
+MIFARE_ACL_READ_BASE_KEY = settings.mifare_acl_read_base_key
-MIFARE_ACL_WRITE_BASE_KEY = os.getenv("MIFARE_ACL_WRITE_BASE_KEY")
+MIFARE_ACL_WRITE_BASE_KEY = settings.mifare_acl_write_base_key
 
 # Constants
 MIFARE_APP_ID = "DEAFFE"  # 7 bytes
@@ -332,9 +332,13 @@ class BackgroundScanner:
             time.sleep(5)
 
     def _check_db(self, key):
-        check = checkAccess(key, self.db)
+        if isDoorOpen():
-        if check:
+            closeDoor()
-            openDoor()
+            logger.info("Door closed by key %s", key)
-            logger.info("Access granted!")
         else:
-            logger.error("Access denied!")
+            check = checkAccess(key, self.db)
+            if check:
+                openDoor()
+                logger.info("Access granted!")
+            else:
+                logger.error("Access denied!")

app/services/settings.py

@@ -1,29 +1,70 @@
 import logging
-import os
+from functools import lru_cache
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
 
 logger = logging.getLogger(__name__)
 
 
-def verify_settings():
+class Settings(BaseSettings):
-    card_envs = [
+    model_config = SettingsConfigDict(env_file=".env")
-        "MIFARE_APP_MASTER_KEY",
+
-        "MIFARE_ACL_READ_BASE_KEY",
+    secret_key: str
-        "MIFARE_ACL_WRITE_BASE_KEY",
+    sqlalchemy_database_url: str = "sqlite:///./gatekeeper.db"
-    ]
+    lock_pin: int = 17
-    important_envs = ["SECRET_KEY"]
+    unlock_pin: int = 18
-    other_envs = ["SQLALCHEMY_DATABASE_PATH"]
+    mock_gpio: bool = True
-    for setting in card_envs:
+    alembic_config: str = "./alembic.ini"
-        if (setting not in os.environ or setting == "") and not os.getenv(
+
-            "DISABLE_CARDS"
+    disable_cards: bool = False
-        ):
+    mifare_app_master_key: str | None = None
-            raise ValueError(
+    mifare_acl_read_base_key: str | None = None
-                f"Missing environment variable for scanner start: {setting} \n Run with DISABLE_CARDS env var to disable cards"  # noqa: E501
+    mifare_acl_write_base_key: str | None = None
+
+
+@lru_cache(1)
+def _create_settings():
+    settings = Settings()
+    if not settings.disable_cards:
+        missing = [
+            name.upper()
+            for name in (
+                "mifare_app_master_key",
+                "mifare_acl_read_base_key",
+                "mifare_acl_write_base_key",
             )
-    for setting in important_envs:
+            if not getattr(settings, name)
-        if setting not in os.environ or setting == "":
+        ]
-            raise ValueError(
+        if missing:
-                f"Missing critical environment variable {setting}. Stopping..."
+            logger.critical(
+                "Missing environment variable for scanner start: %s"
+                "Card scanner and related funcionality is disabled!",
+                ", ".join(missing),
             )
-    for setting in other_envs:
+            settings.disable_cards = True
-        if setting not in os.environ:
+    return settings
-            logger.warning(f"Env var {setting} not set. Continuing with defaults.")
+
+
+class _SettingsProxy:
+    _instance: Settings | None = None
+
+    def _load(self) -> Settings:
+        if self._instance is None:
+            self._instance = _create_settings()
+        return self._instance
+
+    def __getattr__(self, name: str):
+        return getattr(self._load(), name)
+
+    def __setattr__(self, name: str, value):
+        if name == "_instance":
+            super().__setattr__(name, value)
+        else:
+            setattr(self._load(), name, value)
+
+    def reset(self) -> None:
+        self._instance = None
+        _create_settings.cache_clear()
+
+
+settings = _SettingsProxy()

flake.nix

@@ -65,6 +65,10 @@
                   buildInputs = (old.buildInputs or []) ++ [ pkgs.pcsclite.dev ];
                   NIX_CFLAGS_COMPILE = "-I${pkgs.pcsclite.dev}/include/PCSC";
                 });
+                lgpio = prev.lgpio.overrideAttrs (old: {
+                  nativeBuildInputs = (old.nativeBuildInputs or []) ++ [ pkgs.swig ];
+                  buildInputs = (old.buildInputs or []) ++ [ pkgs.lgpio ];
+                });
               })
             ]
           )
@@ -93,7 +97,7 @@
               UV_NO_SYNC = "1";
               UV_PYTHON = pythonSet.python.interpreter;
               UV_PYTHON_DOWNLOADS = "never";
-              LD_LIBRARY_PATH = "${lib.getLib pkgs.pcsclite}/lib";
+              LD_LIBRARY_PATH = "${lib.getLib pkgs.pcsclite}/lib:${lib.getLib pkgs.lgpio}/lib";
             };
             shellHook = ''
               unset PYTHONPATH

module.nix

@@ -6,23 +6,42 @@ in
   options = {
     services.gatekeeper = {
       enable = lib.mkEnableOption "Enable the gatekeeper api service.";
-      dotenv = lib.mkOption {
+      envFile = lib.mkOption {
-        type = lib.types.path;
+        type = lib.types.nullOr lib.types.path;
-        description = "The path to a .env file with the keys";
+        description = "The path to a .env file with all the other options";
       };
       db = lib.mkOption {
         type = lib.types.path;
         description = "Where to save the database.";
+        default = "/var/lib/gatekeeper";
+      };
+      mockGpio = lib.mkOption {
+        type = lib.types.str;
+        default = "True";
+        description = "Mock GPIO pins. Has to be a string!";
       };
     };
   };
   config = lib.mkIf cfg.enable {
     users.groups.gatekeeper = {};
+    users.groups.gpio = {};
     users.users.gatekeeper = {
       description = "gatekeeper user";
       group = "gatekeeper";
+      extraGroups = ["gpio"];
       isSystemUser = true;
     };
+    services.udev.extraRules = lib.mkBefore ''
+      KERNEL=="gpiomem", GROUP="gpio", MODE="0660"
+      SUBSYSTEM=="gpio", KERNEL=="gpiochip*", ACTION=="add", PROGRAM="${pkgs.bash}/bin/bash -c '${pkgs.coreutils}/bin/chgrp gpio /dev/%k && chmod 660 /dev/%k && ${pkgs.coreutils}/bin/chgrp -R gpio /sys/class/gpio && ${pkgs.coreutils}/bin/chmod -R g=u /sys/class/gpio'"
+      SUBSYSTEM=="gpio", ACTION=="add", PROGRAM="${pkgs.bash}/bin/bash -c '${pkgs.coreutils}/bin/chgrp -R gpio /sys%p && ${pkgs.coreutils}/bin/chmod -R g=u /sys%p'"
+    '';
+
+    boot.kernelParams = [
+      "iomem=relaxed" # for pigpiod 
+      "strict-devmem=0"
+    ];
+
     services.pcscd = {
       enable = true;
       plugins = [ pkgs.acsccid ];
@@ -43,6 +62,12 @@ in
         RestartSec = "20";
         StateDirectory = "gatekeeper";
         WorkingDirectory = "/var/lib/gatekeeper";
+        EnvironmentFile = cfg.envFile;
+      };
+      environment = {
+        SQLALCHEMY_DATABASE_URL = "sqlite:///${cfg.db}/gatekeeper.db";
+        ALEMBIC_CONFIG = "${self}/alembic.ini";
+        MOCK_GPIO = cfg.mockGpio;
       };
     };
   };

pyproject.toml

@@ -18,6 +18,8 @@ dependencies = [
     "pyscard>=2.3.1",
     "alembic>=1.18.5",
     "ruff>=0.16.0",
+    "pydantic-settings>=2.13.1",
+    'lgpio>=0.2.2.0' #  ; sys_platform == "linux" and platform_machine == "aarch64"',
 ]
 
 [tool.uv.sources]
@@ -26,6 +28,7 @@ python-desfire = { git = "https://github.com/waza-ari/python-desfire" }
 [tool.uv.extra-build-dependencies]
 python-desfire = ["poetry"]
 "pyscard" = ["setuptools"]
+"lgpio" = ["setuptools"]
 
 [tool.setuptools]
 py-modules = ["app"]
@@ -47,4 +50,4 @@ select = [
     # isort
     "I",
 ]
-ignore = ["B008"]
+ignore = ["B008"]

test/conftest.py

@@ -1,3 +1,4 @@
+import os
 from datetime import time
 
 import pytest
@@ -5,6 +6,8 @@ from fastapi.testclient import TestClient
 from sqlalchemy.pool import StaticPool
 from sqlmodel import Session, SQLModel, create_engine
 
+os.environ["SECRET_KEY"] = "ff" * 16
+
 from app.main import app
 from app.model.models import (
     AccessAuthorizationDB,

test/test_services/test_database.py

@@ -9,10 +9,10 @@ def test_create_db_and_tables():
     # This is primarily an integration test
     from sqlalchemy import inspect
 
-    from app.services.database import engine
+    from app.services.database import get_engine
 
     create_db_and_tables()
-    inspector = inspect(engine)
+    inspector = inspect(get_engine())
 
     # Check that tables exist
     tables = inspector.get_table_names()

uv.lock

@@ -624,8 +624,10 @@ source = { virtual = "." }
 dependencies = [
     { name = "alembic" },
     { name = "fastapi", extra = ["standard"] },
+    { name = "lgpio" },
     { name = "poetry" },
     { name = "pwdlib", extra = ["argon2"] },
+    { name = "pydantic-settings" },
     { name = "pyjwt", extra = ["crypto"] },
     { name = "pyscard" },
     { name = "pytest" },
@@ -641,8 +643,10 @@ dependencies = [
 requires-dist = [
     { name = "alembic", specifier = ">=1.18.5" },
     { name = "fastapi", extras = ["standard"], specifier = ">=0.135.3" },
+    { name = "lgpio", specifier = ">=0.2.2.0" },
     { name = "poetry", specifier = ">=2.3.4" },
     { name = "pwdlib", extras = ["argon2"], specifier = ">=0.3.0" },
+    { name = "pydantic-settings", specifier = ">=2.13.1" },
     { name = "pyjwt", extras = ["crypto"], specifier = ">=2.12.1" },
     { name = "pyscard", specifier = ">=2.3.1" },
     { name = "pytest", specifier = ">=9.0.3" },
@@ -842,6 +846,12 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/81/db/e655086b7f3a705df045bf0933bdd9c2f79bb3c97bfef1384598bb79a217/keyring-25.7.0-py3-none-any.whl", hash = "sha256:be4a0b195f149690c166e850609a477c532ddbfbaed96a404d4e43f8d5e2689f", size = 39160, upload-time = "2025-11-16T16:26:08.402Z" },
 ]
 
+[[package]]
+name = "lgpio"
+version = "0.2.2.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/56/33/26ec2e8049eaa2f077bf23a12dc61ca559fbfa7bea0516bf263d657ae275/lgpio-0.2.2.0.tar.gz", hash = "sha256:11372e653b200f76a0b3ef8a23a0735c85ec678a9f8550b9893151ed0f863fff", size = 90087, upload-time = "2024-03-29T21:59:55.901Z" }
+
 [[package]]
 name = "mako"
 version = "1.3.12"