diff --git a/README.md b/README.md
index 5ebe444..3a7411b 100644
--- a/README.md
+++ b/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`
Start dev server `nix run .#dev` or `nix run .#dev -- {args}`
Interactive dev with `nix develop`, then sync deps with `uv sync`
+There is a nix module you can use by importing `inputs.gatekeeper.nixosModules.gatekeeper`
Swagger UI @ http://127.0.0.1:8000/api/v1/docs
OpenApi @ http://127.0.0.1:8000/api/v1/openapi.json
@@ -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
diff --git a/app/__init__.py b/app/__init__.py
index e4c3274..e69de29 100644
--- a/app/__init__.py
+++ b/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)
diff --git a/app/controllers/doorManager.py b/app/controllers/doorManager.py
index 449b964..2d33924 100644
--- a/app/controllers/doorManager.py
+++ b/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")
diff --git a/app/main.py b/app/main.py
index f8312e9..210c59b 100644
--- a/app/main.py
+++ b/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
-
-# ruff: enable[E402]
+from app.services.settings import settings
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())
-
- if not os.getenv("DISABLE_CARDS"):
+ init_controller(
+ 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)
diff --git a/app/services/auth.py b/app/services/auth.py
index 5b54232..d4392db 100644
--- a/app/services/auth.py
+++ b/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
diff --git a/app/services/database.py b/app/services/database.py
index f1adadc..7b19034 100644
--- a/app/services/database.py
+++ b/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):
- SQLModel.metadata.create_all(engine)
+ inspector = inspect(get_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.attributes["sqlalchemy.url"] = SQLALCHEMY_DATABASE_URL
+ alembic_cfg = Config(settings.alembic_config)
+ 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):
diff --git a/app/services/door.py b/app/services/door.py
index c734ee9..c3fb0ee 100644
--- a/app/services/door.py
+++ b/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
- doorIsOpen = True
- logger.info("Still needs gpio out")
+ get_controller().open()
def closeDoor():
- global doorIsOpen
- doorIsOpen = False
- logger.info("Still needs gpio out")
+ get_controller().close()
def isDoorOpen():
- return doorIsOpen
+ return get_controller().is_open()
def decrementOneshot(db: Session, oneshot: OneShotAccess):
diff --git a/app/services/scanner.py b/app/services/scanner.py
index 72e056c..1b5b23b 100644
--- a/app/services/scanner.py
+++ b/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_ACL_READ_BASE_KEY = os.getenv("MIFARE_ACL_READ_BASE_KEY")
-MIFARE_ACL_WRITE_BASE_KEY = os.getenv("MIFARE_ACL_WRITE_BASE_KEY")
+MIFARE_APP_MASTER_KEY = settings.mifare_app_master_key
+MIFARE_ACL_READ_BASE_KEY = settings.mifare_acl_read_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 check:
- openDoor()
- logger.info("Access granted!")
+ if isDoorOpen():
+ closeDoor()
+ logger.info("Door closed by key %s", key)
else:
- logger.error("Access denied!")
+ check = checkAccess(key, self.db)
+ if check:
+ openDoor()
+ logger.info("Access granted!")
+ else:
+ logger.error("Access denied!")
diff --git a/app/services/settings.py b/app/services/settings.py
index 85a31c2..e22c766 100644
--- a/app/services/settings.py
+++ b/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():
- card_envs = [
- "MIFARE_APP_MASTER_KEY",
- "MIFARE_ACL_READ_BASE_KEY",
- "MIFARE_ACL_WRITE_BASE_KEY",
- ]
- important_envs = ["SECRET_KEY"]
- other_envs = ["SQLALCHEMY_DATABASE_PATH"]
- for setting in card_envs:
- if (setting not in os.environ or setting == "") and not os.getenv(
- "DISABLE_CARDS"
- ):
- raise ValueError(
- f"Missing environment variable for scanner start: {setting} \n Run with DISABLE_CARDS env var to disable cards" # noqa: E501
+class Settings(BaseSettings):
+ model_config = SettingsConfigDict(env_file=".env")
+
+ secret_key: str
+ sqlalchemy_database_url: str = "sqlite:///./gatekeeper.db"
+ lock_pin: int = 17
+ unlock_pin: int = 18
+ mock_gpio: bool = True
+ alembic_config: str = "./alembic.ini"
+
+ disable_cards: bool = False
+ mifare_app_master_key: str | None = None
+ mifare_acl_read_base_key: str | None = None
+ 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 setting not in os.environ or setting == "":
- raise ValueError(
- f"Missing critical environment variable {setting}. Stopping..."
+ if not getattr(settings, name)
+ ]
+ if missing:
+ logger.critical(
+ "Missing environment variable for scanner start: %s"
+ "Card scanner and related funcionality is disabled!",
+ ", ".join(missing),
)
- for setting in other_envs:
- if setting not in os.environ:
- logger.warning(f"Env var {setting} not set. Continuing with defaults.")
+ settings.disable_cards = True
+ return settings
+
+
+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()
diff --git a/flake.nix b/flake.nix
index 22e6445..f9ba513 100644
--- a/flake.nix
+++ b/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
diff --git a/module.nix b/module.nix
index b761f80..aa37337 100644
--- a/module.nix
+++ b/module.nix
@@ -6,23 +6,42 @@ in
options = {
services.gatekeeper = {
enable = lib.mkEnableOption "Enable the gatekeeper api service.";
- dotenv = lib.mkOption {
- type = lib.types.path;
- description = "The path to a .env file with the keys";
+ envFile = lib.mkOption {
+ type = lib.types.nullOr lib.types.path;
+ 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;
};
};
};
diff --git a/pyproject.toml b/pyproject.toml
index d911fe8..15bc111 100644
--- a/pyproject.toml
+++ b/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"]
\ No newline at end of file
+ignore = ["B008"]
diff --git a/test/conftest.py b/test/conftest.py
index 1a4bf40..8f12ee3 100644
--- a/test/conftest.py
+++ b/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,
diff --git a/test/test_services/test_database.py b/test/test_services/test_database.py
index 5030922..2b55570 100644
--- a/test/test_services/test_database.py
+++ b/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()
diff --git a/uv.lock b/uv.lock
index caa56d5..1f3154c 100644
--- a/uv.lock
+++ b/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"