Move caldav_handler to caldav.py

Make the sleep interruptable
Add debug endpoints for caldav - gonna make a new endpoint for them i think...

.gitea/workflows/pr_check.yml

@@ -0,0 +1,33 @@
+name: Pull-Request Vibe Check
+on:
+  pull_request:
+    types: [ opened, synchronize, reopened ]
+
+jobs:
+  ci-test:
+    runs-on: node-current
+    steps:
+      - name: Setup PRBot
+        uses: imgurbot12/prbot@v2
+        with:
+          username:  ${{ secrets.prbot_username }}
+          api_token: ${{ secrets.prbot_token }}
+
+      - uses: actions/checkout@v6
+      - name: Set up UV
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
+
+      - run: |
+          apt update
+          apt install libpcsclite-dev python3-dev -y
+      
+      - name: Install deps
+        run: uv sync --locked --all-extras --dev
+
+      - name: Run ruff check
+        run: uv run ruff check --output-format=github | sed 's|/workspace/malobeo/gatekeeper/||' | prbot
+        continue-on-error: true
+
+      - name: Post PR Feedback
+        if: always()
+        run: prbot commit

.gitea/workflows/python.yml

@@ -0,0 +1,21 @@
+name: Test python action
+
+on: [push]
+
+jobs:
+  test:
+    runs-on: node-current
+    steps:
+      - uses: actions/checkout@v6
+      - name: Set up UV
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
+
+      - run: |
+          apt update
+          apt install libpcsclite-dev python3-dev -y
+      
+      - name: Install deps
+        run: uv sync --locked --all-extras --dev
+
+      - name: Run tests
+        run: uv run pytest test

.gitignore

@@ -1,2 +1,5 @@
 __pycache__
 gatekeeper.db
+.env
+.coverage
+result

.vscode/settings.json

@@ -0,0 +1,11 @@
+{
+    "[python]": {
+        "editor.formatOnSave": true,
+        "editor.defaultFormatter": "charliermarsh.ruff"
+    },
+    "python.testing.pytestArgs": [
+        "test"
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
+}

README.md

@@ -1,15 +1,47 @@
 ## Gatekeeper - Door access system
-Status: WIP - not prod ready
+#### Status: "WORKING" - Base functionality is there, mayor issues
 
-Dev with `nix develop`   
+Start prod server `nix run`<br>
-start dev server `uv run fastapi dev`   
+Start dev server `nix run .#dev` or  `nix run .#dev -- {args}`<br>
-Swagger UI @ http://127.0.0.1:8000/docs  
+Interactive dev with `nix develop`, then sync deps with `uv sync`<br>
-start prod server `uv run fastapi run`
+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>
 
-Issues:
+#### Config:
-   - `nix run` currently broken
+
-   - no auth
+Nixos config for the card reader:
-   - no door state
+```
-   - no door operations
+services.pcscd = {
-   - card system is dummy until I get hardware
+  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:
+   - cards can only unlock, not lock
+   - documentation missing
+   - 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
+   - inconsistent logging (request logging?)
+   - rate limiting maybe
+   - pretty sure the controllers are doing too much stuff
+   - hardware call tests or sth

alembic.ini

@@ -0,0 +1,149 @@
+# A generic, single database configuration.
+
+[alembic]
+# path to migration scripts.
+# this is typically a path given in POSIX (e.g. forward slashes)
+# format, relative to the token %(here)s which refers to the location of this
+# ini file
+script_location = %(here)s/alembic
+
+# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
+# Uncomment the line below if you want the files to be prepended with date and time
+# see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file
+# for all available tokens
+# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s
+# Or organize into date-based subdirectories (requires recursive_version_locations = true)
+# file_template = %%(year)d/%%(month).2d/%%(day).2d_%%(hour).2d%%(minute).2d_%%(second).2d_%%(rev)s_%%(slug)s
+
+# sys.path path, will be prepended to sys.path if present.
+# defaults to the current working directory.  for multiple paths, the path separator
+# is defined by "path_separator" below.
+prepend_sys_path = .
+
+
+# timezone to use when rendering the date within the migration file
+# as well as the filename.
+# If specified, requires the tzdata library which can be installed by adding
+# `alembic[tz]` to the pip requirements.
+# string value is passed to ZoneInfo()
+# leave blank for localtime
+# timezone =
+
+# max length of characters to apply to the "slug" field
+# truncate_slug_length = 40
+
+# set to 'true' to run the environment during
+# the 'revision' command, regardless of autogenerate
+# revision_environment = false
+
+# set to 'true' to allow .pyc and .pyo files without
+# a source .py file to be detected as revisions in the
+# versions/ directory
+# sourceless = false
+
+# version location specification; This defaults
+# to <script_location>/versions.  When using multiple version
+# directories, initial revisions must be specified with --version-path.
+# The path separator used here should be the separator specified by "path_separator"
+# below.
+# version_locations = %(here)s/bar:%(here)s/bat:%(here)s/alembic/versions
+
+# path_separator; This indicates what character is used to split lists of file
+# paths, including version_locations and prepend_sys_path within configparser
+# files such as alembic.ini.
+# The default rendered in new alembic.ini files is "os", which uses os.pathsep
+# to provide os-dependent path splitting.
+#
+# Note that in order to support legacy alembic.ini files, this default does NOT
+# take place if path_separator is not present in alembic.ini.  If this
+# option is omitted entirely, fallback logic is as follows:
+#
+# 1. Parsing of the version_locations option falls back to using the legacy
+#    "version_path_separator" key, which if absent then falls back to the legacy
+#    behavior of splitting on spaces and/or commas.
+# 2. Parsing of the prepend_sys_path option falls back to the legacy
+#    behavior of splitting on spaces, commas, or colons.
+#
+# Valid values for path_separator are:
+#
+# path_separator = :
+# path_separator = ;
+# path_separator = space
+# path_separator = newline
+#
+# Use os.pathsep. Default configuration used for new projects.
+path_separator = os
+
+# set to 'true' to search source files recursively
+# in each "version_locations" directory
+# new in Alembic version 1.10
+# recursive_version_locations = false
+
+# the output encoding used when revision files
+# are written from script.py.mako
+# output_encoding = utf-8
+
+# database URL.  This is consumed by the user-maintained env.py script only.
+# other means of configuring database URLs may be customized within the env.py
+# file.
+# sqlalchemy.url = sqlite:///./gatekeeper.db
+
+
+[post_write_hooks]
+# post_write_hooks defines scripts or Python functions that are run
+# on newly generated revision scripts.  See the documentation for further
+# detail and examples
+
+# format using "black" - use the console_scripts runner, against the "black" entrypoint
+# hooks = black
+# black.type = console_scripts
+# black.entrypoint = black
+# black.options = -l 79 REVISION_SCRIPT_FILENAME
+
+# lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module
+# hooks = ruff
+# ruff.type = module
+# ruff.module = ruff
+# ruff.options = check --fix REVISION_SCRIPT_FILENAME
+
+# Alternatively, use the exec runner to execute a binary found on your PATH
+# hooks = ruff
+# ruff.type = exec
+# ruff.executable = ruff
+# ruff.options = check --fix REVISION_SCRIPT_FILENAME
+
+# Logging configuration.  This is also consumed by the user-maintained
+# env.py script only.
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARNING
+handlers = console
+qualname =
+
+[logger_sqlalchemy]
+level = WARNING
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s
+datefmt = %H:%M:%S

alembic/README

@@ -0,0 +1 @@
+Generic single-database configuration.

alembic/env.py

@@ -0,0 +1,86 @@
+from logging.config import fileConfig
+
+from sqlalchemy import engine_from_config
+from sqlalchemy import pool
+
+from alembic import context
+
+from app.services.settings import settings
+
+# this is the Alembic Config object, which provides
+# access to the values within the .ini file in use.
+config = context.config
+
+# Interpret the config file for Python logging.
+# This line sets up loggers basically.
+#if config.config_file_name is not None:
+#    fileConfig(config.config_file_name, disable_existing_loggers=False)
+
+# add your model's MetaData object here
+# for 'autogenerate' support
+# from myapp import mymodel
+# target_metadata = mymodel.Base.metadata
+from app.model import models
+
+target_metadata = models.Base.metadata
+
+
+def _database_url() -> str:
+    return config.get_main_option("sqlalchemy.url") or settings.sqlalchemy_database_url
+
+# other values from the config, defined by the needs of env.py,
+# can be acquired:
+# my_important_option = config.get_main_option("my_important_option")
+# ... etc.
+
+
+def run_migrations_offline() -> None:
+    """Run migrations in 'offline' mode.
+
+    This configures the context with just a URL
+    and not an Engine, though an Engine is acceptable
+    here as well.  By skipping the Engine creation
+    we don't even need a DBAPI to be available.
+
+    Calls to context.execute() here emit the given string to the
+    script output.
+
+    """
+    url = _database_url()
+    context.configure(
+        url=url,
+        target_metadata=target_metadata,
+        literal_binds=True,
+        dialect_opts={"paramstyle": "named"},
+    )
+
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def run_migrations_online() -> None:
+    """Run migrations in 'online' mode.
+
+    In this scenario we need to create an Engine
+    and associate a connection with the context.
+
+    """
+    connectable = engine_from_config(
+        {"sqlalchemy.url": _database_url()},
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
+
+    with connectable.connect() as connection:
+        context.configure(
+            connection=connection, target_metadata=target_metadata
+        )
+
+        with context.begin_transaction():
+            context.run_migrations()
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()

alembic/script.py.mako

@@ -0,0 +1,28 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+${imports if imports else ""}
+
+# revision identifiers, used by Alembic.
+revision: str = ${repr(up_revision)}
+down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)}
+branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
+depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    ${downgrades if downgrades else "pass"}

alembic/versions/4a40fa23e086_init.py

@@ -0,0 +1,28 @@
+"""init
+
+Revision ID: 4a40fa23e086
+Revises: 
+Create Date: 2026-07-28 00:07:19.402303
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '4a40fa23e086'
+down_revision: Union[str, Sequence[str], None] = None
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    pass
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    pass

alembic/versions/93e8dda267e5_remove_timetable_access_authorizations.py

@@ -0,0 +1,57 @@
+"""remove timetable access authorizations
+
+Revision ID: 93e8dda267e5
+Revises: 4a40fa23e086
+Create Date: 2026-11-05 15:12:41.908153
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers used by Alembic.
+revision: str = '93e8dda267e5'
+down_revision: Union[str, Sequence[str], None] = '4a40fa23e086'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Drop the timetable table and the authorizations that used it.
+
+    create_db_and_tables() only bootstraps an empty database, so existing
+    deployments keep the timetable table and their type='timetable' rows until
+    this migration is applied with `alembic upgrade head`. Such rows can no
+    longer be read: AccessAuthorizationDB has no timetables relationship
+    anymore, door.checkAccess() ignores the type, and AccessAuthorizationResponse
+    rejects the value, which turns every GET on them into a ResponseValidationError.
+    """
+    # Delete the authorizations before the table so the FK ordering stays clean.
+    op.execute(
+        sa.text(
+            "DELETE FROM accessauthorizationdb WHERE type = 'timetable'"
+        )
+    )
+    op.drop_table('timetable')
+
+
+def downgrade() -> None:
+    """Recreate the timetable table.
+
+    The deleted type='timetable' authorizations and their timetable rows are
+    gone for good; a downgrade only restores the schema.
+    """
+    op.create_table(
+        'timetable',
+        sa.Column('weekday', sa.INTEGER(), nullable=False),
+        sa.Column('starttime', sa.TIME(), nullable=False),
+        sa.Column('duration', sa.INTEGER(), nullable=False),
+        sa.Column('id', sa.INTEGER(), nullable=False),
+        sa.Column('accessauth_id', sa.INTEGER(), nullable=False),
+        sa.PrimaryKeyConstraint('id'),
+        sa.ForeignKeyConstraint(
+            ['accessauth_id'], ['accessauthorizationdb.id'],
+        ),
+    )

app/.env.example

@@ -0,0 +1,11 @@
+MIFARE_APP_MASTER_KEY="7b195adb892073c9e34ebe7e0ca28b2b"
+# 16 bytes AES key
+MIFARE_ACL_READ_BASE_KEY="741a50f2b189cc9f151e0600f50a6e1a"
+# 16 bytes AES key
+MIFARE_ACL_WRITE_BASE_KEY="f1aa99f81cca268de98d422ee0ccb65c"
+# 16 bytes AES key
+SECRET_KEY="8b14d0b447bff7efa24d5019cc59a999786e31f6f865173bbd642bf18de5ad85"
+#Key for oauth
+#THESE ARE TESTING KEYS - DO NOT USE IN PROD
+
+SQLALCHEMY_DATABASE_PATH="./gatekeeper.db"

app/__init__.py

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

app/controllers/aaManager.py

@@ -1,41 +1,74 @@
-from fastapi import APIRouter, Depends, HTTPException
+import logging
+
+from fastapi import APIRouter, Depends, HTTPException, status
 from sqlmodel import Session, select
-from sqlalchemy.orm import selectinload
-from typing import List
 
-from ..model.models import *
+from app.model.models import (
-from ..services.database import engine, get_session, add_and_refresh
+    AccessAuthorizationCreate,
-import uuid as gen_uuid
+    AccessAuthorizationDB,
+    AccessAuthorizationResponse,
+    AccessAuthorizationUpdate,
+    GroupDB,
+    GroupResponse,
+    OneShotAccess,
+)
+from app.services.auth import auth_is_admin
+from app.services.database import add_and_refresh, get_session
+
+logger = logging.getLogger(__name__)
+
+aa_router = APIRouter(prefix="/api/v1/aa", tags=["AccessAuth"])
 
-aa_router = APIRouter(prefix="/aa", tags=["AccessAuth"])
 
 @aa_router.post("/", response_model=AccessAuthorizationResponse)
-def add_accessauth(*, db: Session = Depends(get_session), aa: AccessAuthorizationCreate):
+def add_accessauth(
-    print("Creating accessauth with data: ", aa)
+    *,
-    timetables = [Timetable.model_validate(t) for t in aa.timetables]
+    db: Session = Depends(get_session),
+    aa: AccessAuthorizationCreate,
+    admin: bool = Depends(auth_is_admin),
+):
+    logger.info(f"Creating accessauth with data: {aa}")
+    if aa.oneshot is not None:
+        oneshot = OneShotAccess.model_validate(aa.oneshot)
+    else:
+        oneshot = None
     db_aa = AccessAuthorizationDB(
         name=aa.name,
+        type=aa.type,
         is_active=aa.is_active,
-        timetables=timetables
+        oneshot=oneshot,
     )
     return add_and_refresh(db, db_aa)
 
-@aa_router.get("/", response_model=List[AccessAuthorizationResponse])
+
-def get_all_accessauths(db: Session = Depends(get_session)):
+@aa_router.get("/", response_model=list[AccessAuthorizationResponse])
-    return db.exec(
+def get_all_accessauths(
-        select(AccessAuthorizationDB)
+    db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
-        .options(selectinload(AccessAuthorizationDB.timetables))
+):
-        ).all()
+    return db.exec(select(AccessAuthorizationDB)).all()
+
 
 @aa_router.get("/{aa_id}", response_model=AccessAuthorizationResponse)
-def get_one_accessauth(*, db: Session = Depends(get_session), aa_id: int):
+def get_one_accessauth(
+    *,
+    db: Session = Depends(get_session),
+    aa_id: int,
+    admin: bool = Depends(auth_is_admin),
+):
     db_aa = db.get(AccessAuthorizationDB, aa_id)
     if db_aa is None:
         raise HTTPException(status_code=404, detail="AA not found")
     return db_aa
 
+
 @aa_router.put("/assign/{group_id}/{aa_id}", response_model=GroupResponse)
-def assign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa_id: int):
+def assign_accessauth(
+    *,
+    db: Session = Depends(get_session),
+    group_id: int,
+    aa_id: int,
+    admin: bool = Depends(auth_is_admin),
+):
     db_group = db.get(GroupDB, group_id)
     if db_group is None:
         raise HTTPException(status_code=404, detail="Group not found")
@@ -43,12 +76,21 @@ def assign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa_i
     if db_aa is None:
         raise HTTPException(status_code=404, detail="AA not found")
     if db_aa in db_group.accessauths:
-        raise HTTPException(status_code=200, detail="AA already assigned to group")
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="AA already assigned to group"
+        )
     db_group.accessauths.append(db_aa)
     return add_and_refresh(db, db_group)
 
+
 @aa_router.put("/unassign/{group_id}/{aa_id}", response_model=GroupResponse)
-def unassign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa_id: int):
+def unassign_accessauth(
+    *,
+    db: Session = Depends(get_session),
+    group_id: int,
+    aa_id: int,
+    admin: bool = Depends(auth_is_admin),
+):
     db_group = db.get(GroupDB, group_id)
     if db_group is None:
         raise HTTPException(status_code=404, detail="Group not found")
@@ -56,24 +98,41 @@ def unassign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa
     if db_aa is None:
         raise HTTPException(status_code=404, detail="AA not found")
     if db_aa not in db_group.accessauths:
-        raise HTTPException(status_code=200, detail="AA not assigned to group")
+        raise HTTPException(status_code=404, detail="AA not assigned to group")
     db_group.accessauths.remove(db_aa)
     return add_and_refresh(db, db_group)
 
+
 @aa_router.patch("/{aa_id}", response_model=AccessAuthorizationResponse)
-def change_accessauth(*, db: Session = Depends(get_session), aa_id: int, aa: AccessAuthorizationUpdate):
+def change_accessauth(
+    *,
+    db: Session = Depends(get_session),
+    aa_id: int,
+    aa: AccessAuthorizationUpdate,
+    admin: bool = Depends(auth_is_admin),
+):
     db_aa = db.get(AccessAuthorizationDB, aa_id)
     if db_aa is None:
         raise HTTPException(status_code=404, detail="AccessAuthorization not found")
-    aa_data = aa.dict(exclude_unset=True)
+    aa_data = aa.model_dump(exclude_unset=True, exclude_none=True)
+    if "oneshot" in aa_data and aa_data["oneshot"] is not None:
+        oneshot = OneShotAccess.model_validate(aa_data["oneshot"])
+        db_aa.oneshot = oneshot
+        aa_data.pop("oneshot")
     db_aa.sqlmodel_update(aa_data)
     return add_and_refresh(db, db_aa)
 
+
 @aa_router.delete("/{aa_id}")
-def delete_accessauth(*, db: Session = Depends(get_session), aa_id: int):
+def delete_accessauth(
+    *,
+    db: Session = Depends(get_session),
+    aa_id: int,
+    admin: bool = Depends(auth_is_admin),
+):
     db_aa = db.get(AccessAuthorizationDB, aa_id)
     if db_aa is None:
         raise HTTPException(status_code=404, detail="AccessAuthorization not found")
     db.delete(db_aa)
     db.commit()
-    return {"message": "AccessAuthorization deleted successfully"}
+    return {"message": "AccessAuthorization deleted successfully"}

app/controllers/cardManager.py

@@ -1,39 +1,118 @@
-from fastapi import APIRouter, Depends, HTTPException
+import logging
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.exc import NoResultFound
 from sqlmodel import Session, select
-from typing import List
 
-from ..model.models import Card
+from app.model.models import Card, CardCreate, CardUpdate, GroupDB
-from ..services.database import engine, get_session, add_and_refresh
+from app.services.auth import auth_is_admin
-import uuid as gen_uuid
+from app.services.database import add_and_refresh, get_session
+from app.services.scanner import DeleteCard, WriteNewCard
 
-card_router = APIRouter(prefix="/cards", tags=["Card"])
+logger = logging.getLogger(__name__)
 
-def register_card(group_id: int):
+card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"])
-    uuid = str(gen_uuid.uuid4()) #hier code für mifare registrierung
+
-    card = Card(group_id=group_id, uuid=uuid)
+
+def register_card(cardInput: CardCreate):
+    key, uid = WriteNewCard()
+    if key is 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=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):
+@card_router.post("/", response_model=Card)
-    card = register_card(group_id)
+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()
+            is None
+        )
+    except AssertionError as err:
+        raise HTTPException(
+            status.HTTP_409_CONFLICT, detail="Name already used!"
+        ) from err
+    try:
+        assert (
+            db.exec(
+                select(GroupDB).where(GroupDB.id == cardInput.group_id)
+            ).one_or_none()
+            is not None
+        )
+    except AssertionError as err:
+        raise HTTPException(
+            status.HTTP_404_NOT_FOUND, detail="GroupID not found!"
+        ) from err
+    card = register_card(cardInput)
     return add_and_refresh(db, card)
 
-@card_router.delete("/{card_id}")
+
-def del_card(*, db: Session = Depends(get_session), card_id: int):
+@card_router.delete("/")
-    card = db.get(Card, card_id)
+def del_card(
-    if card is None:
+    *, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
-        raise HTTPException(status_code=404, detail="Card not found")
+):
+    key = DeleteCard()
+    logger.info(key)
+    try:
+        card = db.exec(select(Card).where(Card.key == key)).one()
+    except NoResultFound as err:
+        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. KEY={key}",
+        ) from err
     db.delete(card)
     db.commit()
     return {"message": "Card deleted successfully"}
-##TBH not a big fan of having creation using group_id but deletion using card_id
+
-@card_router.get("/{group_id}", response_model=List[Card])
+
-def get_cards(*, db: Session = Depends(get_session), group_id: int):
+@card_router.get("/{group_id}", response_model=list[Card])
+def get_cards(
+    *,
+    db: Session = Depends(get_session),
+    group_id: int,
+    admin: bool = Depends(auth_is_admin),
+):
     cards = db.exec(select(Card).where(Card.group_id == group_id)).all()
     return cards
 
 
-#TODO:
+@card_router.patch("/{card_id}", response_model=Card)
-# -Split Authorisations + Cards
+def update_card(
-# -Deactivation
+    card_id: int,
-# -Deleting
+    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 as err:
+            raise HTTPException(
+                status.HTTP_404_NOT_FOUND, detail="GroupID not found!"
+            ) from err
+    db_card.sqlmodel_update(card_data)
+    return add_and_refresh(db, db_card)

app/controllers/debugManager.py

@@ -0,0 +1,90 @@
+import logging
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import exc
+from sqlmodel import Session, select
+
+from app.model.models import Card
+from app.services.auth import auth_is_admin
+from app.services.caldav import get_caldav_handler
+from app.services.database import add_and_refresh, get_session
+
+logger = logging.getLogger(__name__)
+
+debug_router = APIRouter(
+    prefix="/api/v1/debug",
+    tags=["Debug items - maybe dont show this in UI"],
+    dependencies=[Depends(auth_is_admin)],
+)
+
+
+@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,
+        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_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 as err:
+        raise HTTPException(
+            status_code=500, detail="No card with that id found."
+        ) from err
+    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)):
+    logger.info("Debug Setting: Getting cards.")
+    cards = db.exec(select(Card)).all()
+    print(cards)
+    out = []
+    for i in cards:
+        out.append({i.key: i.group.name})
+    return out
+
+
+@debug_router.get("/calGetActiveEvents")
+def cal_get_active_events(db: Session = Depends(get_session)):
+    handler = get_caldav_handler()
+    if handler is None:
+        raise HTTPException(503, "Caldav handler not initialized")
+    return handler.get_active_events()
+
+
+@debug_router.get("/calGetFutureEvents")
+def cal_get_future_events(db: Session = Depends(get_session)):
+    handler = get_caldav_handler()
+    if handler is None:
+        raise HTTPException(503, "Caldav handler not initialized")
+    return handler.get_future_events()
+
+
+@debug_router.get("/calGetActiveGroups")
+def cal_get_active_groups(db: Session = Depends(get_session)):
+    handler = get_caldav_handler()
+    if handler is None:
+        raise HTTPException(503, "Caldav handler not initialized")
+    return handler.get_active_groups()

app/controllers/doorManager.py

@@ -0,0 +1,32 @@
+from fastapi import APIRouter, Depends
+from sqlmodel import Session
+
+import app.services.door as doorService
+from app.services.auth import auth_is_admin
+from app.services.database import get_session
+
+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()
+
+
+@door_router.put("/close")
+def close_door(
+    db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
+):
+    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")
+def test_access(input: str, db: Session = Depends(get_session)):
+    return doorService.checkAccess(input, db=db)

app/controllers/groupManager.py

@@ -1,27 +1,47 @@
-from fastapi import APIRouter, HTTPException, Depends
+from fastapi import APIRouter, Depends, HTTPException, status
 from sqlmodel import Session, select
-from typing import List
 
-from ..model.models import GroupDB, GroupResponse, GroupCreate
+from ..model.models import GroupCreate, GroupDB, GroupResponse
-from ..services.database import engine, get_session, add_and_refresh
+from ..services.auth import auth_is_admin
+from ..services.database import add_and_refresh, get_session
 
-group_router = APIRouter(prefix="/groups", tags=["Group"])
+group_router = APIRouter(prefix="/api/v1/groups", tags=["Group"])
 
-@group_router.get("/", response_model=List[GroupResponse])
+
-def get_groups(*, db: Session = Depends(get_session)):
+@group_router.get("/", response_model=list[GroupResponse])
+def get_groups(
+    *, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
+):
     groups = db.exec(select(GroupDB)).all()
     return groups
 
+
 @group_router.post("/", response_model=GroupResponse)
-def create_group(*, db: Session = Depends(get_session), group: GroupCreate):
+def create_group(
+    *,
+    db: Session = Depends(get_session),
+    group: GroupCreate,
+    admin: bool = Depends(auth_is_admin),
+):
     db_group = GroupDB.model_validate(group)
+    group = db.exec(select(GroupDB).where(GroupDB.name == db_group.name)).first()
+    if group is not None:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="Group already exists!"
+        )
     return add_and_refresh(db, db_group)
 
+
 @group_router.delete("/{group_id}")
-def delete_group(*, db: Session = Depends(get_session), group_id: int):
+def delete_group(
+    *,
+    db: Session = Depends(get_session),
+    group_id: int,
+    admin: bool = Depends(auth_is_admin),
+):
     db_group = db.get(GroupDB, group_id)
     if db_group is None:
         raise HTTPException(status_code=404, detail="Group not found")
     db.delete(db_group)
     db.commit()
-    return {"message": "Group deleted successfully"}
+    return {"message": "Group deleted successfully"}

app/controllers/userManager.py

@@ -1,47 +1,97 @@
-from fastapi import APIRouter, HTTPException, Depends
+import logging
+
+from fastapi import APIRouter, Depends, HTTPException, status
 from sqlmodel import Session, select
-from typing import List
 
-from ..model.models import UserResponse, UserCreate, UserDB, UserUpdate
+from app.model.models import UserCreate, UserDB, UserResponse, UserUpdate
-from ..services.database import engine, get_session, add_and_refresh
+from app.services.auth import auth_is_admin, get_password_hash
+from app.services.auth import get_current_user as auth_user
+from app.services.database import add_and_refresh, get_session
 
-user_router = APIRouter(tags=["Users"])
+logger = logging.getLogger(__name__)
 
-@user_router.post("/users/", response_model=UserResponse)
+user_router = APIRouter(tags=["Users"], prefix="/api/v1/users")
-def create_user(*, db: Session = Depends(get_session), user: UserCreate):
+
-    print("creating user with data ", user)
+
-    db_user = UserDB.model_validate(user)
+@user_router.post("/", response_model=UserResponse)
+def create_user(
+    *,
+    db: Session = Depends(get_session),
+    user: UserCreate,
+    admin: bool = Depends(auth_is_admin),
+):
+    hashed_password = {"passwordhash": get_password_hash(user.password)}
+    try:
+        assert (
+            db.exec(select(UserDB).where(UserDB.name == user.name)).one_or_none()
+            is None
+        )
+    except AssertionError as err:
+        raise HTTPException(
+            status.HTTP_409_CONFLICT, detail="Name already used!"
+        ) from err
+    db_user = UserDB.model_validate(user, update=hashed_password)
     return add_and_refresh(db, db_user)
 
-@user_router.get("/users/", response_model=List[UserResponse])
+
-def read_users(*, db: Session = Depends(get_session)):
+@user_router.get("/", response_model=list[UserResponse])
+def read_users(
+    *, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
+):
     users = db.exec(select(UserDB)).all()
     return users
 
-@user_router.get("/users/{user_id}", response_model=UserResponse)
+
-def read_user(*, db: Session = Depends(get_session), user_id: int):
+@user_router.get("/current", response_model=UserResponse)
+def get_current_user(
+    db: Session = Depends(get_session), user: UserDB = Depends(auth_user)
+):
+    return user
+
+
+@user_router.get("/{user_id}", response_model=UserResponse)
+def read_user(
+    *,
+    db: Session = Depends(get_session),
+    user_id: int,
+    admin: bool = Depends(auth_is_admin),
+):
     db_user = db.get(UserDB, user_id)
     if db_user is None:
         raise HTTPException(status_code=404, detail="User not found")
     return db_user
 
-@user_router.patch("/users/{user_id}", response_model=UserResponse)
+
-def update_user(*, db: Session = Depends(get_session), user_id: int, user: UserUpdate):
+@user_router.patch("/{user_id}", response_model=UserResponse)
+def update_user(
+    *,
+    db: Session = Depends(get_session),
+    user_id: int,
+    user: UserUpdate,
+    admin: bool = Depends(auth_is_admin),
+):
     db_user = db.get(UserDB, user_id)
     if db_user is None:
         raise HTTPException(status_code=404, detail="User not found")
     user_data = user.model_dump(exclude_unset=True)
-    db_user.sqlmodel_update(user_data)
+    hashed_password = {}
-    db.add(db_user)
+    if "password" in user_data:
-    db.commit()
+        password = user_data["password"]
-    db.refresh(db_user)
+        hashed_password = {"passwordhash": get_password_hash(password)}
-    return db_user
+    db_user.sqlmodel_update(user_data, update=hashed_password)
+    return add_and_refresh(db, db_user)
 
-@user_router.delete("/users/{user_id}")
+
-def delete_user(*, db: Session = Depends(get_session), user_id: int):
+@user_router.delete("/{user_id}")
+def delete_user(
+    *,
+    db: Session = Depends(get_session),
+    user_id: int,
+    admin: bool = Depends(auth_is_admin),
+):
     db_user = db.get(UserDB, user_id)
     if db_user is None:
         raise HTTPException(status_code=404, detail="User not found")
     db.delete(db_user)
     db.commit()
-    return {"message": "User deleted successfully"}
+    return {"message": "User deleted successfully"}

app/main.py

@@ -1,14 +1,78 @@
+import logging
+from contextlib import asynccontextmanager
+
 from fastapi import FastAPI
-from .controllers import userManager, cardManager, groupManager, aaManager
+from fastapi.middleware.cors import CORSMiddleware
-from .services.database import create_db_and_tables
+from fastapi.security import OAuth2PasswordBearer
 
-app = FastAPI()
+from app.controllers import (
-@app.on_event("startup")
+    aaManager,
-def on_startup():
+    cardManager,
+    debugManager,
+    doorManager,
+    groupManager,
+    userManager,
+)
+from app.services.auth import create_first_user, token_router
+from app.services.caldav import init_caldav_handler, stop_caldav_handler
+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 settings
+
+logger = logging.getLogger(__name__)
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+scanner = BackgroundScanner(db=get_db_session())
+logging.basicConfig(level=logging.INFO)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
     create_db_and_tables()
-    print("Database created and tables initialized.")
+    create_first_user(db=get_db_session())
+    init_controller(
+        DoorController(
+            mqtt_host=settings.mqtt_host,
+            mqtt_port=settings.mqtt_port,
+            mock_factory=settings.mock,
+        ),
+    )
+    if not settings.disable_cards:
+        scanner.start()
 
+    init_caldav_handler()
+
+    logger.info("-" * 63)
+    logger.info("---- Documentation is at http://127.0.0.1:8000/api/v1/docs ----")
+    logger.info("-" * 63)
+    yield
+    stop_caldav_handler()
+    # scanner.stop()
+
+
+app = FastAPI(
+    lifespan=lifespan, docs_url="/api/v1/docs", openapi_url="/api/v1/openapi.json"
+)
+
+origins = [
+    "http://127.0.0.1",
+    "http://localhost",
+    "http://localhost:8080",
+]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["GET", "PUT", "POST", "DELETE", "PATCH"],
+    allow_headers=["*"],
+)
+
+app.include_router(token_router)
 app.include_router(userManager.user_router)
 app.include_router(groupManager.group_router)
 app.include_router(cardManager.card_router)
-app.include_router(aaManager.aa_router)
+app.include_router(aaManager.aa_router)
+app.include_router(doorManager.door_router)
+app.include_router(debugManager.debug_router)

app/model/models.py

@@ -1,24 +1,60 @@
-from sqlmodel import Field, Relationship, Session, SQLModel
+from datetime import UTC, datetime
-from typing import List
+from typing import Literal
+
+from pydantic import model_validator
+from sqlalchemy import DateTime, TypeDecorator
+from sqlmodel import Field, Relationship, SQLModel
+
 
 class Base(SQLModel):
     pass
 
+
+#### Extra Column Types
+class UTCDateTime(TypeDecorator):
+    """Modified datetime for timezones
+    Input time gets read as given tz or local tz, then converted to UTC
+    Output everything as UTC
+    """
+
+    impl = DateTime(timezone=True)
+    LOCAL_TIMEZONE = datetime.now(UTC).astimezone().tzinfo
+    cache_ok = True
+
+    def process_bind_param(self, value: datetime | None, dialect) -> datetime | None:
+        if value is None:
+            return None
+        if value.tzinfo is None:
+            value = value.astimezone(self.LOCAL_TIMEZONE)
+        return value.astimezone(UTC)
+
+    def process_result_value(self, value: datetime | None, dialect) -> datetime | None:
+        if value is None:
+            return None
+        if value.tzinfo is None:
+            return value.replace(tzinfo=UTC)
+        return value.astimezone(UTC)
+
+
 #### User
 class UserBase(Base):
-    name: str = Field(index=True)
+    name: str = Field(index=True, unique=True)
     email: str | None = None
     is_admin: bool = False
 
+
 class UserResponse(UserBase):
     id: int
 
+
 class UserCreate(UserBase):
     password: str
 
+
 class UserDB(UserBase, table=True):
     id: int | None = Field(default=None, primary_key=True)
-    password: str
+    passwordhash: str
+
 
 class UserUpdate(Base):
     name: str | None = None
@@ -26,68 +62,123 @@ class UserUpdate(Base):
     is_admin: bool | None = None
     password: str | None = None
 
+
 #### Special
 class AaGroupLink(Base, table=True):
-    group_id: int | None = Field(default=None, foreign_key="groupdb.id", primary_key=True)
+    group_id: int | None = Field(
-    accessauth_id: int | None = Field(default=None, foreign_key="accessauthorizationdb.id", primary_key=True)
+        default=None, foreign_key="groupdb.id", primary_key=True
+    )
+    accessauth_id: int | None = Field(
+        default=None, foreign_key="accessauthorizationdb.id", primary_key=True
+    )
+
+
+#### Token
+class Token(Base):
+    access_token: str
+    token_type: str
+
+
+class TokenData(Base):
+    username: str | None = None
+
 
 #### Group
 class GroupBase(Base):
     name: str = Field(index=True, unique=True)
 
+
 class GroupCreate(GroupBase):
     pass
 
+
 class GroupDB(GroupBase, table=True):
     id: int | None = Field(default=None, primary_key=True)
-    cards: List["Card"] = Relationship(back_populates="group")
+    cards: list["Card"] = Relationship(back_populates="group")
-    accessauths: List["AccessAuthorizationDB"] = Relationship(back_populates="groups", link_model=AaGroupLink)
+    accessauths: list["AccessAuthorizationDB"] = Relationship(
+        back_populates="groups", link_model=AaGroupLink
+    )
+
 
 class GroupResponse(GroupBase):
     id: int
-    cards: List["Card"] | None
+    cards: list["Card"] | None
-    accessauths: List["AccessAuthorizationDB"] | None
+    accessauths: list["AccessAuthorizationDB"] | None
+
 
 #### AccessAuthorization
 class AccessAuthorizationBase(Base):
     name: str = Field(index=True)
+    type: Literal["oneshot", "somefuturespec"]
     is_active: bool
 
+
 class AccessAuthorizationDB(AccessAuthorizationBase, table=True):
     id: int | None = Field(default=None, primary_key=True)
-    groups: List["GroupDB"] = Relationship(back_populates="accessauths", link_model=AaGroupLink)
+    type: str
-    timetables: List["Timetable"] = Relationship(back_populates="accessauth", cascade_delete=True)
+    groups: list["GroupDB"] = Relationship(
+        back_populates="accessauths", link_model=AaGroupLink
+    )
+    oneshot: "OneShotAccess" = Relationship(
+        back_populates="accessauth", cascade_delete=True
+    )
+
+
+class OneShotAccessBase(Base):
+    uses: int = 1
+    ends_at: datetime = Field(sa_type=UTCDateTime)
+
+
+class OneShotAccess(OneShotAccessBase, table=True):
+    id: int | None = Field(default=None, primary_key=True)
+    accessauth_id: int = Field(default=None, foreign_key="accessauthorizationdb.id")
+    accessauth: AccessAuthorizationDB = Relationship(back_populates="oneshot")
+
 
 class AccessAuthorizationCreate(AccessAuthorizationBase):
-    timetables: List["TimetableCreate"]
+    oneshot: OneShotAccessBase | None = None
+
+    @model_validator(mode="after")
+    def check_type(self):
+        if self.type == "oneshot":
+            if not self.oneshot:
+                raise ValueError("oneshot auths require a oneshot object")
+        elif self.type == "somefuturespec":
+            raise ValueError("somefuturespec is not jet implemented. Please do not use")
+        return self
+
 
 class AccessAuthorizationResponse(AccessAuthorizationBase):
     id: int
-    timetables: List["Timetable"]
+    oneshot: OneShotAccessBase | None = None
-    groups: List["GroupDB"]
+    groups: list["GroupDB"]
+
 
 class AccessAuthorizationUpdate(Base):
     name: str | None = None
+    type: Literal["oneshot", "somefuturespec"] | None = None
     is_active: bool | None = None
-    timetables: List["TimetableCreate"] | None = None
+    oneshot: OneShotAccessBase | None = None
-    
+
 
 #### 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 TimetableBase(Base):
-    weekday: int = Field(le=7, ge=1)
-    starttime: str
-    duration: int = Field(gt=0, lt=1440)
 
-class Timetable(TimetableBase, table=True):
+class CardCreate(Base):
-    id: int | None = Field(default=None, primary_key=True)
+    name: str = Field(unique=True, max_length=32)
-    accessauth_id: int = Field(default=None, foreign_key="accessauthorizationdb.id")
+    enabled: bool = True
-    accessauth: AccessAuthorizationDB = Relationship(back_populates="timetables")
+    group_id: int
 
-class TimetableCreate(TimetableBase):
+
-    pass
+class CardUpdate(Base):
+    name: str | None = None
+    enabled: bool | None = None
+    group_id: int | None = None

app/services/auth.py

@@ -0,0 +1,135 @@
+import logging
+import secrets
+import string
+from datetime import UTC, datetime, timedelta
+from typing import Annotated
+
+import jwt
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+from jwt.exceptions import InvalidTokenError
+from pwdlib import PasswordHash
+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 = settings.secret_key
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 120
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/token")
+
+token_router = APIRouter(tags=["Token"], prefix="/api/v1")
+
+password_hash = PasswordHash.recommended()
+
+
+def verify_password(plain_password, hashed_password):
+    return password_hash.verify(plain_password, hashed_password)
+
+
+def get_password_hash(password):
+    return password_hash.hash(password)
+
+
+def get_user(db, username: str):
+    user = db.exec(select(UserDB).where(UserDB.name == username)).first()
+    return user
+
+
+def authenticate_user(db, username: str, password: str):
+    user = get_user(db, username)
+    if not user:
+        return False
+    if not verify_password(password, user.passwordhash):
+        return False
+    return user
+
+
+def create_access_token(data: dict, expires_delta: timedelta | None = None):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.now(UTC) + expires_delta
+    else:
+        expire = datetime.now(UTC) + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+
+def get_current_user(
+    token: Annotated[str, Depends(oauth2_scheme)],
+    db: Session = Depends(get_session),
+):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+        token_data = TokenData(username=username)
+    except InvalidTokenError as err:
+        raise credentials_exception from err
+    user = get_user(db, username=token_data.username)
+    if user is None:
+        raise credentials_exception
+    return user
+
+
+def auth_is_admin(
+    token: str = Depends(oauth2_scheme),
+    db: Session = Depends(get_session),
+):
+    user = get_current_user(token=token, db=db)
+    if not user.is_admin:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Not authorized to perform this action",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    return True
+
+
+def create_first_user(db: Session):
+    logger.info("Checking for admin user")
+    admin_user = db.exec(select(UserDB)).first()
+    if admin_user is None:
+        password = "".join(secrets.choice(string.digits) for i in range(8))
+        logger.info(f"Creating first admin user with password: {password}")
+        user = UserDB(
+            name="admin", passwordhash=get_password_hash(password), is_admin=True
+        )
+        return add_and_refresh(db, user)
+    logger.info(f"Admin user already exists: {admin_user.name}")
+
+
+@token_router.post("/token")
+def login_for_access_token(
+    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
+    db: Session = Depends(get_session),
+) -> Token:
+    user = authenticate_user(db, form_data.username, form_data.password)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or pw",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    access_token = create_access_token(
+        data={"sub": user.name}, expires_delta=access_token_expires
+    )
+    return Token(access_token=access_token, token_type="bearer")
+
+
+@token_router.get("/test/login")
+def test_login(current_user: Annotated[UserDB, Depends(get_current_user)]) -> UserDB:
+    return current_user

app/services/caldav.py

@@ -0,0 +1,186 @@
+import logging
+import threading
+from datetime import UTC, date, datetime, timedelta
+
+from caldav import get_calendars
+
+from app.services.settings import settings
+
+logger = logging.getLogger(__name__)
+
+
+class CaldavEventCache:
+    def __init__(self, url, username, password, calendarNames):
+        self.url = url
+        self.username = username
+        self.password = password
+        self.calendarNames = calendarNames
+        self.events = []
+        self.is_running = False
+        self.thread: threading.Thread | None = None
+        self._stopped = threading.Event()
+
+    def start(self):
+        if self.is_running:
+            logger.error("CaldavEventCache already running")
+            return
+        self.is_running = True
+        self._stopped.clear()
+        self.thread = threading.Thread(target=self._refresh_loop, daemon=True)
+        self.thread.start()
+        logger.info("CaldavEventCache started")
+
+    def stop(self):
+        self.is_running = False
+        self._stopped.set()
+        if self.thread:
+            self.thread.join()
+        logger.info("CaldavEventCache stopped")
+
+    def _refresh_loop(self):
+        while self.is_running:
+            try:
+                self.refresh()
+            except Exception as e:
+                logger.error(f"Error updating CaldavEventCache: {e}", exc_info=True)
+            self._stopped.wait(settings.caldav_refresh_interval)
+
+    def refresh(self, daysBack=1, daysAhead=7):
+        calendars = get_calendars(
+            url=self.url,
+            username=self.username,
+            password=self.password,
+            calendar_name=self.calendarNames,
+        )
+
+        cached = []
+        now = datetime.now(UTC)
+        for cal in calendars:
+            my_events = cal.search(
+                event=True,
+                start=now - timedelta(days=daysBack),
+                end=now + timedelta(days=daysAhead),
+                expand=True,
+            )
+            for event in my_events:
+                component = event.component
+                start, end = component.start, component.end
+                all_day = isinstance(start, date) and not isinstance(start, datetime)
+
+                if not all_day:
+                    # Treat timezone-naive event times as system-local time.
+                    if start.tzinfo is None:
+                        start = start.astimezone()
+                    if end.tzinfo is None:
+                        end = end.astimezone()
+
+                cached.append(
+                    {
+                        "uid": str(component.get("uid", "")),
+                        "summary": str(component.get("summary", "")),
+                        "groups": str(component.get("description", "")).splitlines(),
+                        "start": start,
+                        "end": end,
+                        "all_day": all_day,
+                    }
+                )
+
+        # dont cache empty array.
+        # it most probably means there was error during caldav connection
+        # TODO: handle error properly
+        if len(cached) == 0:
+            return
+
+        self.events = cached
+
+    def get_active_events(self, active_time=None):
+        active_time = active_time or datetime.now().astimezone()
+        if active_time.tzinfo is None:
+            active_time = active_time.astimezone()
+
+        today = active_time.date()
+        active = []
+
+        for event in self.events:
+            start, end = event["start"], event["end"]
+
+            if event["all_day"]:
+                # All-day event end dates are exclusive.
+                is_active = start <= today < end
+            else:
+                is_active = start <= active_time < end
+
+            if is_active:
+                active.append(event)
+
+        return active
+
+    def get_future_events(self, active_time=None):
+        active_time = active_time or datetime.now().astimezone()
+        if active_time.tzinfo is None:
+            active_time = active_time.astimezone()
+
+        today = active_time.date()
+        future = []
+
+        for event in self.events:
+            start = event["start"]
+
+            if event["all_day"]:  # noqa: SIM108
+                # All-day event end dates are exclusive.
+                is_active = start <= today
+            else:
+                is_active = start <= active_time
+
+            if is_active:
+                future.append(event)
+
+        return future
+
+    def get_active_groups(self, active_time=None):
+        active_events = self.get_active_events(active_time)
+
+        # combine all groups into a list of lists
+        active_groups = []
+        for active_event in active_events:
+            active_groups.append(active_event["groups"])
+
+        # flatten the lists, make every group unique
+        return set(group for groups in active_groups for group in groups)
+
+    def is_group_active(self, group_name, active_time=None):
+        return group_name in self.get_active_groups(active_time)
+
+
+_caldav_handler: CaldavEventCache | None = None
+
+
+def init_caldav_handler() -> CaldavEventCache | None:
+    """Build and start the event cache from settings.
+    Returns None when caldav handling is disabled
+    """
+    global _caldav_handler
+    if settings.disable_caldav:
+        logger.info("Caldav handling disabled")
+        return None
+
+    handler = CaldavEventCache(
+        url=settings.caldav_url,
+        username=settings.caldav_user,
+        password=settings.caldav_password,
+        calendarNames=settings.caldav_calendars,
+    )
+    handler.start()
+    _caldav_handler = handler
+    return handler
+
+
+def get_caldav_handler() -> CaldavEventCache | None:
+    return _caldav_handler
+
+
+def stop_caldav_handler() -> None:
+    global _caldav_handler
+    if _caldav_handler is not None:
+        _caldav_handler.stop()
+        _caldav_handler = None

app/services/database.py

@@ -1,20 +1,90 @@
-from sqlmodel import create_engine, SQLModel, Session
+import logging
 
-from ..model.models import Base
+from sqlalchemy import inspect
+from sqlmodel import Session, SQLModel, create_engine
 
-SQLALCHEMY_DATABASE_URL = "sqlite:///./gatekeeper.db"
+from app.services.settings import settings
+
+logger = logging.getLogger(__name__)
+
+# the lifespan startup (create_db_and_tables, create_first_user),
+# get_db_session(), the module level scanner session and
+# the get_session() FastAPI dependency resolve their engine
+# through get_engine(). That makes configure_engine() the single
+# injection point for tests
+_engine = None
+_engine_is_default = False
+
+
+def _build_engine_from_settings():
+    return create_engine(
+        settings.sqlalchemy_database_url, connect_args={"check_same_thread": False}
+    )
+
+
+def get_engine():
+    """Return the engine, building it from settings on first use."""
+    global _engine, _engine_is_default
+    if _engine is None:
+        _engine = _build_engine_from_settings()
+        _engine_is_default = True
+    return _engine
+
+
+def configure_engine(engine) -> None:
+    """Install *engine* as the engine used by this module from now on.
+
+    The replaced engine is not closed because the scanner (or other long-lived session)
+    could still hold connections
+    """
+    global _engine, _engine_is_default
+    _engine = engine
+    _engine_is_default = False
+
+
+def reset_engine() -> None:
+    """Drop a configured engine and rebuild from settings on the next call.
+
+    Only an engine this module built itself is closed.
+    """
+    global _engine, _engine_is_default
+    if _engine is not None and _engine_is_default:
+        _engine.dispose()
+    _engine = None
+    _engine_is_default = False
 
-engine = create_engine(SQLALCHEMY_DATABASE_URL)
 
 def create_db_and_tables():
-    SQLModel.metadata.create_all(engine)
+    engine = get_engine()
+    existing_tables = inspect(engine).get_table_names()
+    if not existing_tables:
+        SQLModel.metadata.create_all(engine)
+        from alembic.config import Config
+
+        from alembic import command
+
+        alembic_cfg = Config(settings.alembic_config)
+        alembic_cfg.set_main_option("sqlalchemy.url", str(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 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(get_engine())
+
+
 def add_and_refresh(db: Session, obj):
     db.add(obj)
     db.commit()
     db.refresh(obj)
-    return obj
+    return obj

app/services/door.py

@@ -0,0 +1,133 @@
+import logging
+from datetime import datetime
+
+from sqlalchemy import exc
+from sqlmodel import select
+
+from app.model.models import Card, OneShotAccess
+from app.services.caldav import get_caldav_handler
+from app.services.database import Session, add_and_refresh
+
+logger = logging.getLogger(__name__)
+
+
+# See: https://github.com/technyon/nuki_hub#gpio-lock-control-optional
+# TODO: add sensor pin
+class DoorController:
+    def __init__(
+        self,
+        mqtt_host: str = "localhost",
+        mqtt_port: int = 1883,
+        mock_factory: bool = False,
+    ):
+        self._is_open: bool = False
+        self._mqtt = None
+        self._mqtt_topic = "nukihub/lock/action"
+        self._mock = mock_factory
+
+        if not mock_factory:
+            import paho.mqtt.client as mqtt
+
+            self._mqtt = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
+            self._mqtt.connect(
+                mqtt_host, mqtt_port
+            )  # TODO: add login with username+pw, tls
+            self._mqtt.loop_start()
+            logger.info("Mqtt client connect to %s:%s", mqtt_host, mqtt_port)
+
+        logger.info(
+            "DoorController started. topic=%s mock=%s",
+            self._mqtt_topic,
+            mock_factory,
+        )
+
+    def open(self):
+        if self._mock:
+            self._is_open = True
+            logger.info("Door unlocked.[MOCK]")
+            return
+
+        self._mqtt.publish(self._mqtt_topic, "unlock")
+        self._is_open = True
+        logger.info("Door unlocked!")
+
+    def close(self):
+        if self._mock:
+            self._is_open = False
+            logger.info("Door locked.[MOCK]")
+            return
+
+        self._mqtt.publish(self._mqtt_topic, "lock")
+        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():
+    get_controller().open()
+
+
+def closeDoor():
+    get_controller().close()
+
+
+def isDoorOpen():
+    return get_controller().is_open()
+
+
+def decrementOneshot(db: Session, oneshot: OneShotAccess):
+    data = oneshot.model_dump()
+    if data["uses"] > 0:
+        data["uses"] = data["uses"] - 1
+        oneshot.sqlmodel_update(oneshot, update=data)
+        add_and_refresh(db, oneshot)
+
+
+def checkAccess(key: str, db: Session):
+    try:
+        card = db.exec(select(Card).where(Card.key == key)).one()
+        if not card.enabled:
+            logger.info("Card Inactive!")
+            return False
+
+        caldav_handler = get_caldav_handler()
+        if caldav_handler and caldav_handler.is_group_active(card.group.name):
+            logger.info("Access Valid!")
+            return True
+
+        current_time = datetime.now().astimezone()
+        for auth in card.group.accessauths:
+            logger.info(f"checking auth: {auth.name}")
+            if not auth.is_active:
+                logger.info("AA inactive!")
+                continue
+            if auth.type == "oneshot":
+                logger.info(f"  oneshot auth found: {auth.oneshot}")
+                ends_at = auth.oneshot.ends_at
+                if ends_at.tzinfo is None:
+                    ends_at = ends_at.astimezone()
+                if current_time < ends_at and auth.oneshot.uses > 0:
+                    decrementOneshot(db, auth.oneshot)
+                    return True
+        logger.info("No more auths found")
+        return False
+    except exc.NoResultFound as err:
+        raise Exception(
+            "No Access with that key found, this might be a db error"
+        ) from err

app/services/scanner.py

@@ -0,0 +1,342 @@
+import logging
+import secrets
+import threading
+import time
+
+from desfire import (
+    DESFire,
+    DESFireKey,
+    PCSCDevice,
+    diversify_key,
+    get_list,
+    to_hex_string,
+)
+from desfire.enums import (
+    DESFireCommunicationMode,
+    DESFireFileType,
+    DESFireKeySettings,
+    DESFireKeyType,
+)
+from desfire.schemas import FilePermissions, FileSettings, KeySettings
+from fastapi import HTTPException, status
+from smartcard.CardRequest import CardRequest
+from smartcard.CardType import AnyCardType
+from smartcard.Exceptions import CardRequestTimeoutException
+
+from app.services.door import checkAccess, closeDoor, isDoorOpen, openDoor
+from app.services.settings import settings
+
+logger = logging.getLogger(__name__)
+
+# ENV vars
+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
+MIFARE_ACL_READ_BASE_KEY_ID = 0x1
+MIFARE_ACL_WRITE_BASE_KEY_ID = 0x2
+MIFARE_SYS_ID = "FF0000"  # 3 bytes, can essentially be anything
+MIFARE_ENCRYPTED_FILE_ID = 0x1
+
+
+def checkForKey():
+    if MIFARE_APP_MASTER_KEY is None:
+        logger.critical("NO MASTER KEY LOADED")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="No key loaded! Check application.",
+        )
+
+
+def getCardService(timeout: int = 10):
+    cardtype = AnyCardType()
+    cardrequest = CardRequest(timeout=timeout, cardType=cardtype)
+    cardservice = cardrequest.waitforcard()
+    cardservice.connection.connect()
+    return cardservice
+
+
+def readFileOnCard(desfire: DESFire):
+    if not MIFARE_ACL_READ_BASE_KEY:
+        logger.critical("MIFARE_ACL_READ_BASE_KEY not found! Reading skipped!")
+        return
+    # create keys
+    # desfire = DESFire(PCSCDevice(cardservice.connection.component))
+    aes_keysettings = KeySettings(key_type=DESFireKeyType.DF_KEY_AES)
+    keysettings = desfire.get_key_setting()
+    desKey = DESFireKey(keysettings, "00" * 8)
+
+    # Get real UID
+    desfire.authenticate(0x0, desKey)
+    # To get the uid you have to auth with an empty (default) key
+    uid = desfire.get_real_uid()
+    applications = desfire.get_application_ids()
+    try:
+        assert len(applications) == 1
+        assert applications[0] == get_list(MIFARE_APP_ID)
+    except AssertionError:
+        logger.error("No application found!")
+        time.sleep(4)
+        return
+    # Then use the key derivation with that uid, the appid, the sysid
+    diversification_data = (
+        [0x01] + uid + get_list(MIFARE_APP_ID) + get_list(MIFARE_SYS_ID)
+    )
+    read_div_key_bytes = diversify_key(
+        get_list(MIFARE_ACL_READ_BASE_KEY), diversification_data, pad_to_32=False
+    )
+
+    # Log in with derived read key
+    logger.debug("Start auth")
+    aes_app_read_key = DESFireKey(aes_keysettings, read_div_key_bytes)
+    desfire.select_application(MIFARE_APP_ID)
+
+    desfire.authenticate(MIFARE_ACL_READ_BASE_KEY_ID, aes_app_read_key)
+
+    logger.debug(f"Read data from {MIFARE_ENCRYPTED_FILE_ID}")
+    file_data = desfire.get_file_settings(MIFARE_ENCRYPTED_FILE_ID)
+    rdata = desfire.read_file_data(MIFARE_ENCRYPTED_FILE_ID, file_data)
+    # convert list of int to str
+    rdata = to_hex_string(rdata).replace(" ", "").lower()
+    logger.debug(f"Data on card: {rdata}")
+    return rdata
+
+
+def DeleteCard():
+    try:
+        checkForKey()
+        from app.main import scanner as scannerThread
+
+        scannerThread.stop()
+        cardservice = getCardService(15)
+
+        # Create Desfire object
+        desfire = DESFire(PCSCDevice(cardservice.connection.component))
+
+        rdata = readFileOnCard(desfire=desfire)
+
+        # Create Key objects
+        aes_keysettings = KeySettings(key_type=DESFireKeyType.DF_KEY_AES)
+        des_keysettings = KeySettings(key_type=DESFireKeyType.DF_KEY_2K3DES)
+        desKey = DESFireKey(des_keysettings, "00" * 8)
+        aes_master_key = DESFireKey(aes_keysettings, MIFARE_APP_MASTER_KEY)
+        aes_null_key = DESFireKey(aes_keysettings, "00" * 16)
+
+        desfire.select_application(0x0)
+
+        try:
+            try:
+                logger.debug("Auth1")
+                desfire.authenticate(0x0, aes_master_key)
+            except (Exception, AssertionError):
+                logger.debug("Auth2")
+                desfire.authenticate(0x0, aes_null_key)
+        except (Exception, AssertionError):
+            logger.debug("Auth3")
+            desfire.authenticate(0x0, desKey)
+
+        applications = desfire.get_application_ids()
+        logger.debug(f"Applications: {applications}")
+        if len(applications) == 0:
+            raise HTTPException(
+                status_code=status.HTTP_410_GONE, detail="No applications on card"
+            )
+
+        desfire.select_application(MIFARE_APP_ID)
+        desfire.authenticate(0x0, aes_master_key)
+
+        try:
+            desfire.delete_application(MIFARE_APP_ID)
+            logger.info("App deleted!")
+        except Exception:
+            pass
+        scannerThread.start()
+        return rdata
+    except (Exception, AssertionError) as err:
+        logger.error(f"Error in deletion function: {err}", exc_info=True)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error: {err}"
+        ) from err
+
+
+def WriteNewCard():
+    try:
+        checkForKey()
+        from app.main import scanner as scannerThread
+
+        scannerThread.stop()
+
+        cardservice = getCardService(20)
+
+        desfire = DESFire(PCSCDevice(cardservice.connection.component))
+
+        # Create Key objects
+        aes_keysettings = KeySettings(key_type=DESFireKeyType.DF_KEY_AES)
+        aes_null_key = DESFireKey(aes_keysettings, "00" * 16)
+        aes_master_key = DESFireKey(aes_keysettings, MIFARE_APP_MASTER_KEY)
+        desKey = DESFireKey(desfire.get_key_setting(), "00" * 8)
+
+        # Authenticate with default DES key
+        logger.debug("Authenticating with default DES key...")
+        desfire.authenticate(0x0, desKey)
+
+        # get uid
+        uid = desfire.get_real_uid()
+
+        # Set default key
+        logger.debug("Setting default key...")
+        desfire.change_default_key(aes_null_key, 0x0)
+
+        # Create application
+        logger.debug("Creating application...")
+        app_settings = KeySettings(
+            settings=[
+                DESFireKeySettings.KS_ALLOW_CHANGE_MK,
+                DESFireKeySettings.KS_LISTING_WITHOUT_MK,
+                DESFireKeySettings.KS_CREATE_DELETE_WITHOUT_MK,
+                DESFireKeySettings.KS_CONFIGURATION_CHANGEABLE,
+            ],
+            key_type=DESFireKeyType.DF_KEY_AES,
+        )
+        desfire.create_application(MIFARE_APP_ID, app_settings, 4)
+
+        # Verify application creation
+        applications = desfire.get_application_ids()
+        assert len(applications) == 1
+        assert applications[0] == get_list(MIFARE_APP_ID)
+        logger.debug("  - Application created successfully.")
+
+        # Select application
+        desfire.select_application(MIFARE_APP_ID)
+
+        # recreate key object
+        desfire.authenticate(0x0, aes_null_key)
+        desfire.change_key(0x0, aes_null_key, aes_master_key, 0x1)
+
+        logger.debug("new key auth")
+        desfire.authenticate(0x0, aes_master_key)
+
+        aes_null_key = DESFireKey(aes_keysettings, "00" * 16)
+
+        # generate div data
+        diversification_data = (
+            [0x01] + uid + get_list(MIFARE_APP_ID) + get_list(MIFARE_SYS_ID)
+        )
+        read_div_key_bytes = diversify_key(
+            get_list(MIFARE_ACL_READ_BASE_KEY), diversification_data, pad_to_32=False
+        )
+        write_div_key_bytes = diversify_key(
+            get_list(MIFARE_ACL_WRITE_BASE_KEY), diversification_data, pad_to_32=False
+        )
+
+        logger.debug("Changing file read key...")
+        aes_file_read_key = DESFireKey(aes_keysettings, read_div_key_bytes)
+        desfire.change_key(
+            MIFARE_ACL_READ_BASE_KEY_ID, aes_null_key, aes_file_read_key, 0x1
+        )
+
+        logger.debug("Changing file write key...")
+        aes_file_write_key = DESFireKey(aes_keysettings, write_div_key_bytes)
+        desfire.change_key(
+            MIFARE_ACL_WRITE_BASE_KEY_ID, aes_null_key, aes_file_write_key, 0x1
+        )
+
+        logger.debug("Create encrypted file containing key...")
+        file_settings = FileSettings(
+            file_size=16,
+            encryption=DESFireCommunicationMode.ENCRYPTED,
+            permissions=FilePermissions(
+                read_key=MIFARE_ACL_READ_BASE_KEY_ID,
+                write_key=MIFARE_ACL_WRITE_BASE_KEY_ID,
+            ),
+            file_type=DESFireFileType.MDFT_STANDARD_DATA_FILE,
+        )
+        desfire.create_standard_file(MIFARE_ENCRYPTED_FILE_ID, file_settings)
+        file_data = desfire.get_file_settings(MIFARE_ENCRYPTED_FILE_ID)
+
+        logger.debug("Writing UID to encrypted file...")
+        key = secrets.token_hex(16)
+        desfire.write_file_data(
+            MIFARE_ENCRYPTED_FILE_ID, 0x0, file_data.encryption, get_list(key)
+        )
+
+        logger.debug("Reading from encrypted file...")
+        rdata = desfire.read_file_data(MIFARE_ENCRYPTED_FILE_ID, file_data)
+        assert rdata == get_list(key)
+        logger.debug("  - Data written successfully.")
+        scannerThread.start()
+        return key, to_hex_string(data=uid, separator=":")
+
+    except Exception as err:
+        logger.error(f"Error in write function: {err}", exc_info=True)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error: {err}"
+        ) from err
+
+
+class BackgroundScanner:
+    def __init__(self, db):
+        self.db = db
+        self.is_running = False
+        self.thread: threading.Thread | None = None
+
+    def start(self):
+        if self.is_running:
+            logger.error("Scanner already running")
+            return
+        self.is_running = True
+        self.thread = threading.Thread(target=self._scan_loop, daemon=True)
+        self.thread.start()
+        logger.info("Scanner started")
+
+    def stop(self):
+        self.is_running = False
+        if self.thread:
+            self.thread.join()
+        logger.info("Scanner stopped")
+
+    def _scan_loop(self):
+        while self.is_running:
+            try:
+                card_content = self._read_card()
+                if card_content:
+                    self._check_db(card_content)
+                    time.sleep(5)
+                    logger.debug("READY after success")
+                else:
+                    time.sleep(0.1)
+                    logger.debug("READY after timout")
+
+            except Exception as e:
+                logger.error(f"Error in scan function: {e}", exc_info=True)
+                time.sleep(6)
+
+    def _read_card(self):
+        time.sleep(0.5)
+        try:
+            cardservice = getCardService(3)
+        except CardRequestTimeoutException:
+            logger.debug("No tag detected within the timeout.")
+            return
+        # Create Desfire object
+        desfire = DESFire(PCSCDevice(cardservice.connection.component))
+        try:
+            rdata = readFileOnCard(desfire=desfire)
+            return rdata
+        except Exception as e:
+            logger.error(f"something went wrong: {e}")
+            time.sleep(5)
+
+    def _check_db(self, key):
+        if isDoorOpen():
+            closeDoor()
+            logger.info("Door closed by key %s", key)
+        else:
+            check = checkAccess(key, self.db)
+            if check:
+                openDoor()
+                logger.info("Access granted!")
+            else:
+                logger.error("Access denied!")

app/services/settings.py

@@ -0,0 +1,97 @@
+import logging
+from functools import lru_cache
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+logger = logging.getLogger(__name__)
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file=".env")
+
+    secret_key: str
+    sqlalchemy_database_url: str = "sqlite:///./gatekeeper.db"
+    mqtt_host: str = "localhost"
+    mqtt_port: int = 1883
+    mock: 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
+
+    disable_caldav: bool = True
+    caldav_url: str | None = None
+    caldav_user: str | None = None
+    caldav_password: str | None = None
+    caldav_calendars: list[str] | None = None
+    caldav_refresh_interval: int | 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",
+            )
+            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),
+            )
+            settings.disable_cards = True
+
+    if not settings.disable_caldav:
+        missing = [
+            name.upper()
+            for name in (
+                "caldav_url",
+                "caldav_user",
+                "caldav_password",
+                "caldav_calendars",
+                "caldav_refresh_interval",
+            )
+            if not getattr(settings, name)
+        ]
+        if missing:
+            logger.critical(
+                "Missing environment variable for caldav handling: %s "
+                "Caldav handling and related funcionality is disabled!",
+                ", ".join(missing),
+            )
+            settings.disable_caldav = 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()

flake.lock

@@ -2,11 +2,11 @@
   "nodes": {
     "nixpkgs": {
       "locked": {
-        "lastModified": 1775710090,
+        "lastModified": 1778869304,
-        "narHash": "sha256-ar3rofg+awPB8QXDaFJhJ2jJhu+KqN/PRCXeyuXR76E=",
+        "narHash": "sha256-30sZNZoA1cqF5JNO9fVX+wgiQYjB7HJqqJ4ztCDeBZE=",
         "owner": "nixos",
         "repo": "nixpkgs",
-        "rev": "4c1018dae018162ec878d42fec712642d214fdfa",
+        "rev": "d233902339c02a9c334e7e593de68855ad26c4cb",
         "type": "github"
       },
       "original": {
@@ -29,11 +29,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1773870109,
+        "lastModified": 1776659114,
-        "narHash": "sha256-ZoTdqZP03DcdoyxvpFHCAek4bkPUTUPUF3oCCgc3dP4=",
+        "narHash": "sha256-qapCOQmR++yZSY43dzrp3wCrkOTLpod+ONtJWBk6iKU=",
         "owner": "pyproject-nix",
         "repo": "build-system-pkgs",
-        "rev": "b6e74f433b02fa4b8a7965ee24680f4867e2926f",
+        "rev": "ffaa2161dd5d63e0e94591f86b54fc239660fb2e",
         "type": "github"
       },
       "original": {
@@ -49,11 +49,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1776120154,
+        "lastModified": 1778901413,
-        "narHash": "sha256-mtIBTmVKzyoFYoAGdd8Cd7iswFna9YQVyjObZLXPO64=",
+        "narHash": "sha256-GSKXTAnFqRAMlZkJrIPcQMYf+lpMr66K3i60mB9STvc=",
         "owner": "pyproject-nix",
         "repo": "pyproject.nix",
-        "rev": "29dc4e9960d2b7f122b52b155e0e8f87cd5c5c08",
+        "rev": "a228447c3e179d477c1b6246ef3efa8cfe3c469a",
         "type": "github"
       },
       "original": {
@@ -80,11 +80,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1776114780,
+        "lastModified": 1779269674,
-        "narHash": "sha256-aYUgp40qkY7oNSm+G9A/woNYP+eDeFM0bYckfmxEUiY=",
+        "narHash": "sha256-P1LHCRdYpdtHAEzuEsNHrI6d9mVPl5a2fyFDZGHNVbI=",
         "owner": "pyproject-nix",
         "repo": "uv2nix",
-        "rev": "73ff87a3e489b07b9cf842f917963a9e40d49225",
+        "rev": "69aec536f6d1acc415ed2e20299312802aba98c6",
         "type": "github"
       },
       "original": {

flake.nix

@@ -25,6 +25,7 @@
 
   outputs =
     {
+      self,
       nixpkgs,
       pyproject-nix,
       uv2nix,
@@ -58,6 +59,17 @@
             lib.composeManyExtensions [
               pyproject-build-systems.overlays.wheel
               overlay
+              (final: prev: {
+                pyscard = prev.pyscard.overrideAttrs (old: {
+                  nativeBuildInputs = (old.nativeBuildInputs or []) ++ [ pkgs.swig pkgs.pkg-config];
+                  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 ];
+                });
+              })
             ]
           )
       );
@@ -69,18 +81,23 @@
         let
           pkgs = nixpkgs.legacyPackages.${system};
           pythonSet = pythonSets.${system}.overrideScope editableOverlay;
-          virtualenv = pythonSet.mkVirtualEnv "gatekeeper" workspace.deps.all;
+          virtualenv = (pythonSet.mkVirtualEnv "gatekeeper" workspace.deps.all).overrideAttrs(old: { venvIgnoreCollisions = ["*"];});
         in
         {
           default = pkgs.mkShell {
             packages = [
               virtualenv
               pkgs.uv
+              pkgs.pcsclite
+              pkgs.pcsclite.dev
+              pkgs.swig
+              pkgs.pkg-config
             ];
             env = {
               UV_NO_SYNC = "1";
               UV_PYTHON = pythonSet.python.interpreter;
               UV_PYTHON_DOWNLOADS = "never";
+              LD_LIBRARY_PATH = "${lib.getLib pkgs.pcsclite}/lib:${lib.getLib pkgs.lgpio}/lib";
             };
             shellHook = ''
               unset PYTHONPATH
@@ -91,7 +108,26 @@
       );
 
       packages = forAllSystems (system: {
-        default = pythonSets.${system}.mkVirtualEnv "gatekeeper" workspace.deps.default;
+        default = (pythonSets.${system}.mkVirtualEnv "gatekeeper" workspace.deps.default).overrideAttrs(old: { venvIgnoreCollisions = ["*"];});
       });
+
+      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 = { config, pkgs, lib, ... }@args:
+        import ./module.nix (args // { self = self; system = pkgs.system; });
     };
 }

module.nix

@@ -0,0 +1,74 @@
+{config, pkgs, lib, self, system, ... }:
+let
+  cfg = config.services.gatekeeper;
+in
+{
+  options = {
+    services.gatekeeper = {
+      enable = lib.mkEnableOption "Enable the gatekeeper api service.";
+      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";
+      };
+      mock = lib.mkOption {
+        type = lib.types.str;
+        default = "False";
+        description = "Mock the doorcontroller. 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 ];
+    };
+    networking.firewall.allowedTCPPorts = [ 8000 ];
+    systemd.services.gatekeeper = {
+      description = "Runs the gatekeeper api";
+      script = ''
+        export LD_LIBRARY_PATH="${lib.getLib pkgs.pcsclite}/lib''${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}"
+        exec ${self.packages.${system}.default}/bin/uvicorn --host 0.0.0.0 --port 8000 --app-dir ${self} app.main:app
+      '';
+      after = [ "network.target" ];
+      wantedBy = ["multi-user.target"];
+      
+      serviceConfig = {
+        User = "gatekeeper";
+        Restart = "on-failure";
+        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 = cfg.mock;
+      };
+    };
+  };
+}

plan.txt

@@ -42,7 +42,14 @@ Plan:
         Erstelle gruppe
         Erstelle AA mit Timeslot
         Registriere karte -> gruppe
-        
+    
+    CRUD
+        create(post)
+        read(get)[all items]
+        read(get)[single item]
+        update(put/patch)
+        delete(delete)
+
     zeitplan
         Hier bin ich nicht sicher, ich denke an cron style für wiederholende dinge aber das kann nur zeitpunkte und keine blöcke.
         Villeicht ne liste von cron zeiten [ "0 16 * * 2" "0 18 * * 2" ](Wäre dienstag 16-18 uhr) - Ist aber warscheinlich schwer zu parsen

pyproject.toml

@@ -9,6 +9,22 @@ dependencies = [
     "sqlmodel>=0.0.38",
     "poetry>=2.3.4",
     "python-desfire",
+    "pyjwt[crypto]>=2.12.1",
+    "pwdlib[argon2]>=0.3.0",
+    "requests>=2.33.1",
+    "setuptools>=82.0.1",
+    "pyscard>=2.3.1",
+    "alembic>=1.18.5",
+    "pydantic-settings>=2.13.1",
+    "paho-mqtt>=2.1.0",
+    "caldav>=3.7.0",
+]
+[dependency-groups]
+dev = [
+    "ruff>=0.16.0",
+    "pytest>=9.0.3",
+    "pytest-cov>=7.1.0",
+    "jedi-language-server>=0.47.0",
 ]
 
 [tool.uv.sources]
@@ -16,3 +32,39 @@ 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"]
+
+[tool.ruff]
+exclude = ["alembic"]
+[tool.ruff.lint]
+select = [
+    # pycodestyle
+    "E",
+    # Pyflakes
+    "F",
+    # pyupgrade
+    "UP",
+    # flake8-bugbear
+    "B",
+    # flake8-simplify
+    "SIM",
+    # isort
+    "I",
+]
+ignore = ["B008"]
+
+[tool.pytest.ini_options]
+testpaths = ["test"]
+
+[tool.coverage.run]
+source = ["app"]
+relative_files = true
+
+[tool.coverage.report]
+fail_under = 65
+show_missing = false
+skip_covered = true

test/conftest.py

@@ -0,0 +1,237 @@
+import os
+from datetime import datetime, timedelta
+from time import tzset
+
+import pytest
+from alembic.config import Config
+from alembic.script import ScriptDirectory
+from fastapi.testclient import TestClient
+from sqlalchemy import inspect
+from sqlalchemy.pool import StaticPool
+from sqlmodel import Session, SQLModel, create_engine, select
+
+# TEST_ENV must cover every Settings field
+# test_settings.py fails with the missing name if it does not.
+TEST_ENV = {
+    "SECRET_KEY": "ff" * 16,
+    "SQLALCHEMY_DATABASE_URL": "sqlite://",
+    "MQTT_HOST": "localhost",
+    "MQTT_PORT": "1883",
+    "MOCK": "True",
+    "ALEMBIC_CONFIG": "./alembic.ini",
+    "DISABLE_CARDS": "True",
+    "MIFARE_APP_MASTER_KEY": "0" * 16,
+    "MIFARE_ACL_READ_BASE_KEY": "1" * 16,
+    "MIFARE_ACL_WRITE_BASE_KEY": "2" * 16,
+    "DISABLE_CALDAV": "True",
+    "CALDAV_URL": "http://caldav.invalid/dav.php",
+    "CALDAV_USER": "test-user",
+    "CALDAV_PASSWORD": "test-password",
+    "CALDAV_CALENDARS": '["test-calendar"]',
+    "CALDAV_REFRESH_INTERVAL": "3600",
+}
+
+
+for _name in TEST_ENV:
+    os.environ.pop(_name, None)
+os.environ.update(TEST_ENV)
+
+import app.services.database as database_service  # noqa: E402
+
+TEST_ADMIN_PASSWORD = "admin123"
+
+engine = create_engine(
+    TEST_ENV["SQLALCHEMY_DATABASE_URL"],
+    connect_args={"check_same_thread": False},
+    poolclass=StaticPool,
+)
+
+# inject custom engine
+database_service.configure_engine(engine)
+
+# import main after the custom engine is injected
+from app.main import app  # noqa: E402
+from app.model.models import (  # noqa: E402
+    AccessAuthorizationDB,
+    Card,
+    GroupDB,
+    OneShotAccess,
+    UserDB,
+)
+from app.services.settings import settings  # noqa: E402
+
+
+def alembic_head() -> str:
+    """Head revision of the migration scripts."""
+    alembic_cfg = Config(settings.alembic_config)
+    return ScriptDirectory.from_config(alembic_cfg).get_current_head()
+
+
+def _seed_alembic_version() -> None:
+    """Mark the test database as already migrated.
+
+    Without a version row create_db_and_tables() treats the database as new and
+    runs an alembic stamp, which builds its own engine from the configured URL
+    instead of using this shared in-memory engine.
+    """
+    with engine.connect() as connection:
+        if "alembic_version" not in inspect(connection).get_table_names():
+            connection.exec_driver_sql(
+                "CREATE TABLE alembic_version (version_num VARCHAR(32) NOT NULL)"
+            )
+        connection.exec_driver_sql(
+            "INSERT OR IGNORE INTO alembic_version (version_num) VALUES (?)",
+            (_HEAD_REVISION,),
+        )
+        connection.commit()
+
+
+def _seed_default_admin() -> None:
+    """Seed the admin user the fixtures log in with.
+    TestClient runs lifespan, that calls create_first_user()
+    which blocks the admin_user fixture. Pre-creating the admin stops that
+    """
+    from app.services.auth import get_password_hash
+
+    with Session(engine) as session:
+        if session.exec(select(UserDB).where(UserDB.name == "admin")).first() is None:
+            session.add(
+                UserDB(
+                    name="admin",
+                    passwordhash=get_password_hash(TEST_ADMIN_PASSWORD),
+                    is_admin=True,
+                )
+            )
+            session.commit()
+
+
+_HEAD_REVISION = alembic_head()
+_seed_alembic_version()
+
+
+@pytest.fixture
+def system_tz(request):
+    """Run a test with the host-local timezone forced to a POSIX TZ string."""
+    previous_tz = os.environ.get("TZ")
+    os.environ["TZ"] = request.param
+    tzset()
+    try:
+        yield request.param
+    finally:
+        if previous_tz is None:
+            os.environ.pop("TZ", None)
+        else:
+            os.environ["TZ"] = previous_tz
+        tzset()
+
+
+@pytest.fixture(autouse=True)
+def _installed_engine():
+    """Undo configure_engine() leaks between tests."""
+    previous = database_service.get_engine()
+    yield
+    database_service.configure_engine(previous)
+
+
+@pytest.fixture(scope="function")
+def db_session():
+    """Create a fresh database session for each test."""
+    SQLModel.metadata.create_all(engine)
+    _seed_alembic_version()
+    _seed_default_admin()
+    with Session(engine) as session:
+        yield session
+    SQLModel.metadata.drop_all(engine)
+
+
+@pytest.fixture(scope="function")
+def client(db_session):
+    """Create a test client bound to the test engine.
+
+    No FastAPI dependency override is needed: get_session() resolves its engine
+    through get_engine(), which already points at the test engine.
+    """
+    with TestClient(app) as test_client:
+        yield test_client
+
+
+@pytest.fixture
+def admin_user(db_session):
+    """Get the admin user the seeded database already contains."""
+    admin = db_session.exec(select(UserDB).where(UserDB.name == "admin")).one()
+    db_session.refresh(admin)
+    return admin
+
+
+@pytest.fixture
+def regular_user(db_session):
+    """Create a regular user for testing."""
+    from app.services.auth import get_password_hash
+
+    user = UserDB(
+        name="user", passwordhash=get_password_hash("user123"), is_admin=False
+    )
+    db_session.add(user)
+    db_session.commit()
+    db_session.refresh(user)
+    return user
+
+
+@pytest.fixture
+def auth_headers(client, admin_user):
+    """Get authentication headers for admin user."""
+    response = client.post(
+        "/api/v1/token",
+        data={"username": admin_user.name, "password": TEST_ADMIN_PASSWORD},
+    )
+    token = response.json()["access_token"]
+    return {"Authorization": f"Bearer {token}"}
+
+
+@pytest.fixture
+def user_auth_headers(client, regular_user):
+    """Get authentication headers for regular user."""
+    response = client.post(
+        "/api/v1/token", data={"username": regular_user.name, "password": "user123"}
+    )
+    token = response.json()["access_token"]
+    return {"Authorization": f"Bearer {token}"}
+
+
+@pytest.fixture
+def test_group(db_session):
+    """Create a test group."""
+    group = GroupDB(name="Test Group")
+    db_session.add(group)
+    db_session.commit()
+    db_session.refresh(group)
+    return group
+
+
+@pytest.fixture
+def test_card(db_session, test_group):
+    """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",
+    )
+    db_session.add(card)
+    db_session.commit()
+    db_session.refresh(card)
+    return card
+
+
+@pytest.fixture
+def test_aa_os(db_session):
+    oneshot = OneShotAccess(uses=1, ends_at=datetime.now() + timedelta(days=1))
+    db_session.add(oneshot)
+    aa = AccessAuthorizationDB(name="Test oneshot AA", is_active=True, type="oneshot")
+    aa.oneshot = oneshot
+
+    db_session.add(aa)
+    db_session.commit()
+    db_session.refresh(aa)
+    return aa

test/test_main.py

@@ -0,0 +1,12 @@
+def test_router_includes():
+    """Test that all routers are included in the app."""
+    from app.main import app
+
+    routes = [route.path for route in app.routes]
+
+    # Check that router prefixes are present
+    assert any("/users" in route for route in routes)
+    assert any("/cards" in route for route in routes)
+    assert any("/groups" in route for route in routes)
+    assert any("/aa" in route for route in routes)
+    assert any("/token" in route for route in routes)

test/test_models.py

@@ -0,0 +1,35 @@
+import pytest
+from pydantic import ValidationError
+
+from app.model.models import AccessAuthorizationCreate, UserUpdate
+
+
+def test_user_update_fields_are_optional():
+    """Partial update models must leave unspecified fields as None."""
+    user_update = UserUpdate(name="Updated Name")
+    assert user_update.name == "Updated Name"
+    assert user_update.email is None
+
+
+def test_access_authorization_type_is_validated():
+    """AA validation test"""
+    base = {"name": "AA", "is_active": True}
+
+    with pytest.raises(ValidationError) as exc_info:
+        AccessAuthorizationCreate(type="timetable", **base)
+    assert exc_info.value.errors()[0]["type"] == "literal_error"
+
+    # A oneshot type without its oneshot payload is incomplete.
+    with pytest.raises(ValueError, match="require a oneshot object"):
+        AccessAuthorizationCreate(type="oneshot", **base)
+
+    with pytest.raises(ValueError, match="not jet implemented"):
+        AccessAuthorizationCreate(type="somefuturespec", **base)
+
+    valid = AccessAuthorizationCreate(
+        type="oneshot",
+        name="AA",
+        is_active=True,
+        oneshot={"uses": 1, "ends_at": "2029-07-27"},
+    )
+    assert valid.type == "oneshot"

test/test_services/test_aa_manager.py

@@ -0,0 +1,376 @@
+from datetime import UTC, datetime
+
+import pytest
+from sqlalchemy import text
+from sqlmodel import select
+
+from app.model.models import OneShotAccess
+
+# Berlin's DST rules as a POSIX TZ string: +01:00 in winter, +02:00 in summer.
+BERLIN_TZ = "CET-1CEST,M3.5.0,M10.5.0/3"
+
+
+def _post_oneshot_aa(client, auth_headers, name: str, ends_at: str):
+    """Create a oneshot access authorization and return the response."""
+    return client.post(
+        "/api/v1/aa/",
+        json={
+            "name": name,
+            "type": "oneshot",
+            "is_active": True,
+            "oneshot": {"uses": 1, "ends_at": ends_at},
+        },
+        headers=auth_headers,
+    )
+
+
+def test_create_access_auth_os(client, auth_headers):
+    """Test creating a new access authorization with oneshot type."""
+    aa_data = {
+        "name": "New os_AA",
+        "type": "oneshot",
+        "is_active": True,
+        "oneshot": {"uses": 1, "ends_at": "2029-07-27"},
+    }
+
+    response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers)
+    assert response.status_code == 200
+
+    data = response.json()
+    assert data["name"] == "New os_AA"
+    assert data["type"] == "oneshot"
+    assert data["is_active"] is True
+    assert "id" in data
+    assert data["oneshot"]["uses"] == 1
+
+
+def test_create_wrong_aa_type(client, auth_headers):
+    """Test creating a new access authorization with oneshot type."""
+    aa_data = {
+        "name": "New AA",
+        "type": "wrong",
+        "is_active": True,
+    }
+
+    response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers)
+    assert response.status_code == 422
+
+
+def test_create_timetable_aa_is_rejected(client, auth_headers):
+    """The deprecated timetable type must be refused"""
+    response = client.post(
+        "/api/v1/aa/",
+        json={"name": "Legacy timetable AA", "type": "timetable", "is_active": True},
+        headers=auth_headers,
+    )
+    assert response.status_code == 422
+
+
+def test_update_access_auth_to_timetable_is_rejected(client, auth_headers, test_aa_os):
+    """PATCH must refuse type="timetable" """
+    response = client.patch(
+        f"/api/v1/aa/{test_aa_os.id}",
+        json={"type": "timetable"},
+        headers=auth_headers,
+    )
+    assert response.status_code == 422
+
+    still_fine = client.get(f"/api/v1/aa/{test_aa_os.id}", headers=auth_headers)
+    assert still_fine.status_code == 200
+    assert still_fine.json()["type"] == "oneshot"
+
+
+def test_get_all_access_auths(client, auth_headers, test_aa_os):
+    """Test retrieving all access authorizations."""
+    response = client.get("/api/v1/aa/", headers=auth_headers)
+    assert response.status_code == 200
+
+    aa_list = response.json()
+    assert len(aa_list) >= 1
+    aa_names = [aa["name"] for aa in aa_list]
+    assert test_aa_os.name in aa_names
+
+
+def test_get_access_auth_by_id(client, auth_headers, test_aa_os):
+    """Test retrieving a specific access authorization by ID."""
+    response = client.get(f"/api/v1/aa/{test_aa_os.id}", headers=auth_headers)
+    assert response.status_code == 200
+
+    data = response.json()
+    assert data["id"] == test_aa_os.id
+    assert data["name"] == test_aa_os.name
+
+
+def test_get_nonexistent_access_auth(client, auth_headers):
+    """Test retrieving a non-existent access authorization."""
+    response = client.get("/api/v1/aa/99999", headers=auth_headers)
+    assert response.status_code == 404
+
+
+def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa_os):
+    """Test assigning an access authorization to a group."""
+    response = client.put(
+        f"/api/v1/aa/assign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
+    )
+    assert response.status_code == 200
+
+    data = response.json()
+    assert data["id"] == test_group.id
+    # The AA should now be in the group's accessauths
+    # Note: The response model might not include the full relationship
+
+
+def test_assign_already_assigned_access_auth(
+    client, auth_headers, test_group, test_aa_os
+):
+    """Test assigning an already assigned access authorization."""
+    # First assignment
+    client.put(
+        f"/api/v1/aa/assign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
+    )
+
+    # Second assignment should indicate it's already assigned
+    response = client.put(
+        f"/api/v1/aa/assign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
+    )
+    # According to the code, this returns 409 with "already assigned" message
+    assert response.status_code == 409
+    assert "already assigned" in response.json()["detail"].lower()
+
+
+def test_unassign_access_auth_from_group(client, auth_headers, test_group, test_aa_os):
+    """Test unassigning an access authorization from a group."""
+    # First assign
+    client.put(
+        f"/api/v1/aa/assign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
+    )
+
+    # Then unassign
+    response = client.put(
+        f"/api/v1/aa/unassign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
+    )
+    assert response.status_code == 200
+
+
+def test_unassign_nonexistent_assignment(client, auth_headers, test_group, test_aa_os):
+    """Test unassigning a non-existent assignment."""
+    response = client.put(
+        f"/api/v1/aa/unassign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
+    )
+    assert response.status_code == 404
+
+
+def test_assign_to_nonexistent_group(client, auth_headers, test_aa_os):
+    """Test assigning an AA to a non-existent group."""
+    response = client.put(
+        f"/api/v1/aa/assign/99999/{test_aa_os.id}", headers=auth_headers
+    )
+    assert response.status_code == 404
+
+
+def test_assign_nonexistent_aa(client, auth_headers, test_group):
+    """Test assigning a non-existent AA to a group."""
+    response = client.put(
+        f"/api/v1/aa/assign/{test_group.id}/99999", headers=auth_headers
+    )
+    assert response.status_code == 404
+
+
+def test_update_access_auth(client, auth_headers, test_aa_os):
+    """Test updating an access authorization."""
+    update_data = {"name": "Updated AA", "is_active": False}
+
+    response = client.patch(
+        f"/api/v1/aa/{test_aa_os.id}", json=update_data, headers=auth_headers
+    )
+    assert response.status_code == 200
+
+    data = response.json()
+    assert data["name"] == "Updated AA"
+    assert data["is_active"] is False
+
+
+def test_update_nonexistent_access_auth(client, auth_headers):
+    """Test updating a non-existent access authorization."""
+    update_data = {"name": "Updated"}
+    response = client.patch("/api/v1/aa/99999", json=update_data, headers=auth_headers)
+    assert response.status_code == 404
+
+
+def test_delete_access_auth(client, auth_headers, test_aa_os):
+    """Test deleting an access authorization."""
+    response = client.delete(f"/api/v1/aa/{test_aa_os.id}", headers=auth_headers)
+    assert response.status_code == 200
+    assert "deleted successfully" in response.json()["message"].lower()
+
+    # Verify AA is deleted
+    response = client.get(f"/api/v1/aa/{test_aa_os.id}", headers=auth_headers)
+    assert response.status_code == 404
+
+
+def test_delete_nonexistent_access_auth(client, auth_headers):
+    """Test deleting a non-existent access authorization."""
+    response = client.delete("/api/v1/aa/99999", headers=auth_headers)
+    assert response.status_code == 404
+
+
+def test_aa_os_operations_by_non_admin(client, test_aa_os, user_auth_headers):
+    """Test that non-admin users cannot perform AA operations."""
+    # Try to create an AA
+    response = client.post(
+        "/api/v1/aa/",
+        json={"name": "test", "is_active": True},
+        headers=user_auth_headers,
+    )
+    assert response.status_code == 403
+
+    # Try to get all AAs
+    response = client.get("/api/v1/aa/", headers=user_auth_headers)
+    assert response.status_code == 403
+
+    # Try to assign AA
+    response = client.put(
+        f"/api/v1/aa/assign/1/{test_aa_os.id}", headers=user_auth_headers
+    )
+    assert response.status_code == 403
+
+
+@pytest.mark.parametrize("system_tz", [BERLIN_TZ, "UTC-2"], indirect=True)
+def test_create_access_auth_os_returns_timezone(client, auth_headers, system_tz):
+    """The oneshot end time must come back with a timezone, not as a naive value."""
+    response = _post_oneshot_aa(
+        client, auth_headers, "tz os_AA", "2029-07-27T12:00:00+02:00"
+    )
+    assert response.status_code == 200
+
+    ends_at = response.json()["oneshot"]["ends_at"]
+    parsed = datetime.fromisoformat(ends_at)
+    assert parsed.utcoffset() is not None, f"response has no timezone: {ends_at}"
+    assert parsed == datetime(2029, 7, 27, 10, 0, tzinfo=UTC), (
+        f"the +02:00 offset was not normalised to UTC: {ends_at}"
+    )
+
+
+def test_get_access_auth_os_returns_timezone(client, auth_headers):
+    """Reading the value back from the database keeps it timezone aware."""
+    created = _post_oneshot_aa(
+        client, auth_headers, "tz os_AA", "2029-07-27T12:00:00+02:00"
+    )
+    assert created.status_code == 200
+    aa_id = created.json()["id"]
+
+    for endpoint in (f"/api/v1/aa/{aa_id}", "/api/v1/aa/"):
+        response = client.get(endpoint, headers=auth_headers)
+        assert response.status_code == 200
+        body = response.json()
+        items = body if isinstance(body, list) else [body]
+        auths = [aa for aa in items if aa.get("oneshot")]
+        assert auths, f"no oneshot auth in {endpoint}: {body}"
+        for auth in auths:
+            ends_at = auth["oneshot"]["ends_at"]
+            parsed = datetime.fromisoformat(ends_at)
+            assert parsed.utcoffset() is not None, (
+                f"{endpoint} has no timezone: {ends_at}"
+            )
+            assert parsed == datetime(2029, 7, 27, 10, 0, tzinfo=UTC)
+
+
+def test_oneshot_timezone_is_stored_in_db(client, auth_headers, db_session):
+    """The submitted offset is normalized before the value reaches the database."""
+    response = _post_oneshot_aa(
+        client, auth_headers, "tz os_AA", "2029-07-27T12:00:00+02:00"
+    )
+    assert response.status_code == 200
+
+    db_session.expire_all()
+    stored = db_session.exec(select(OneShotAccess)).one()
+    assert stored.ends_at == datetime(2029, 7, 27, 10, 0, tzinfo=UTC)
+
+    # The column itself must hold the UTC wall clock, not the submitted local time.
+    raw = db_session.exec(text("SELECT ends_at FROM oneshotaccess")).one()[0]
+    assert datetime.fromisoformat(raw) == datetime(2029, 7, 27, 10, 0), (
+        f"unexpected value in the database: {raw}"
+    )
+
+
+@pytest.mark.parametrize("system_tz", [BERLIN_TZ], indirect=True)
+def test_oneshot_without_offset_is_treated_as_local_time(
+    client, auth_headers, db_session, system_tz
+):
+    """A datetime without an offset is wall-clock time on the host, not UTC."""
+    response = _post_oneshot_aa(
+        client, auth_headers, "naive os_AA", "2029-07-27T12:00:00"
+    )
+    assert response.status_code == 200
+
+    # 2029-07-27 is summer in Berlin, so 12:00 local is 10:00 UTC.
+    parsed = datetime.fromisoformat(response.json()["oneshot"]["ends_at"])
+    assert parsed == datetime(2029, 7, 27, 10, 0, tzinfo=UTC)
+
+    db_session.expire_all()
+    stored = db_session.exec(select(OneShotAccess)).one()
+    assert stored.ends_at == datetime(2029, 7, 27, 10, 0, tzinfo=UTC)
+
+
+@pytest.mark.parametrize("system_tz", [BERLIN_TZ], indirect=True)
+def test_oneshot_date_only_is_local_midnight(client, auth_headers, system_tz):
+    """A bare date is local midnight, which is not the same instant as 00:00 UTC."""
+    response = _post_oneshot_aa(client, auth_headers, "date os_AA", "2029-07-27")
+    assert response.status_code == 200
+
+    ends_at = response.json()["oneshot"]["ends_at"]
+    parsed = datetime.fromisoformat(ends_at)
+    assert parsed == datetime(2029, 7, 26, 22, 0, tzinfo=UTC), (
+        f"local midnight was not converted to UTC: {ends_at}"
+    )
+
+
+def test_update_access_auth_os_returns_timezone(client, auth_headers, db_session):
+    """PATCHing the oneshot keeps the timezone in the response and in the DB."""
+    created = _post_oneshot_aa(
+        client, auth_headers, "tz os_AA", "2029-07-27T12:00:00+02:00"
+    )
+    assert created.status_code == 200
+    aa_id = created.json()["id"]
+
+    response = client.patch(
+        f"/api/v1/aa/{aa_id}",
+        json={"oneshot": {"uses": 2, "ends_at": "2030-01-01T05:00:00+05:00"}},
+        headers=auth_headers,
+    )
+    assert response.status_code == 200
+
+    ends_at = response.json()["oneshot"]["ends_at"]
+    parsed = datetime.fromisoformat(ends_at)
+    assert parsed.utcoffset() is not None, f"response has no timezone: {ends_at}"
+    assert parsed == datetime(2030, 1, 1, 0, 0, tzinfo=UTC)
+
+    db_session.expire_all()
+    stored = db_session.exec(select(OneShotAccess)).one()
+    assert stored.ends_at == datetime(2030, 1, 1, 0, 0, tzinfo=UTC)
+
+
+@pytest.mark.parametrize("system_tz", [BERLIN_TZ], indirect=True)
+def test_oneshot_local_time_uses_the_dst_rule_of_its_own_date(
+    client, auth_headers, system_tz
+):
+    """Naive local times must pick the offset in force on that date.
+
+    Without this the host's *current* offset gets applied to every value, so a
+    summer date entered in winter is stored an hour off.
+    """
+    summer = _post_oneshot_aa(
+        client, auth_headers, "summer os_AA", "2029-07-27T12:00:00"
+    )
+    winter = _post_oneshot_aa(
+        client, auth_headers, "winter os_AA", "2029-12-27T12:00:00"
+    )
+    assert summer.status_code == 200 and winter.status_code == 200
+
+    # CEST is UTC+2, CET is UTC+1.
+    assert datetime.fromisoformat(summer.json()["oneshot"]["ends_at"]) == datetime(
+        2029, 7, 27, 10, 0, tzinfo=UTC
+    )
+    assert datetime.fromisoformat(winter.json()["oneshot"]["ends_at"]) == datetime(
+        2029, 12, 27, 11, 0, tzinfo=UTC
+    )

test/test_services/test_auth.py

@@ -0,0 +1,202 @@
+from datetime import timedelta
+
+import pytest
+from fastapi import HTTPException, status
+
+from app.model.models import UserDB
+from app.services.auth import (
+    auth_is_admin,
+    authenticate_user,
+    create_access_token,
+    create_first_user,
+    get_current_user,
+    get_password_hash,
+    get_user,
+    verify_password,
+)
+
+
+def test_password_hashing():
+    """Test password hashing and verification."""
+    password = "test_password_123"
+
+    # Hash password
+    hashed = get_password_hash(password)
+    assert hashed != password
+    assert len(hashed) > 0
+
+    # Verify correct password
+    assert verify_password(password, hashed) is True
+
+    # Verify incorrect password
+    assert verify_password("wrong_password", hashed) is False
+
+
+def test_get_user(db_session):
+    """Test get_user function."""
+    from app.services.auth import get_password_hash
+
+    # Create a user
+    user = UserDB(name="testuser", passwordhash=get_password_hash("password"))
+    db_session.add(user)
+    db_session.commit()
+
+    # Get existing user
+    retrieved_user = get_user(db_session, "testuser")
+    assert retrieved_user is not None
+    assert retrieved_user.name == "testuser"
+
+    # Try to get non-existent user
+    retrieved_user = get_user(db_session, "nonexistent")
+    assert retrieved_user is None
+
+
+def test_authenticate_user(db_session):
+    """Test user authentication."""
+    from app.services.auth import get_password_hash
+
+    # Create a user
+    user = UserDB(name="authuser", passwordhash=get_password_hash("correctpass"))
+    db_session.add(user)
+    db_session.commit()
+
+    # Authenticate with correct credentials
+    authenticated = authenticate_user(db_session, "authuser", "correctpass")
+    assert authenticated is not False
+    assert authenticated.name == "authuser"
+
+    # Authenticate with wrong password
+    authenticated = authenticate_user(db_session, "authuser", "wrongpass")
+    assert authenticated is False
+
+    # Authenticate non-existent user
+    authenticated = authenticate_user(db_session, "nonexistent", "password")
+    assert authenticated is False
+
+
+def test_create_access_token():
+    """Test JWT token creation."""
+    data = {"sub": "testuser"}
+
+    # Create token with default expiration
+    token = create_access_token(data)
+    assert isinstance(token, str)
+    assert len(token) > 0
+
+    # Create token with custom expiration
+    custom_expire = timedelta(hours=1)
+    token = create_access_token(data, expires_delta=custom_expire)
+    assert isinstance(token, str)
+
+
+def test_get_current_user(db_session, admin_user):
+    """Test getting current user from token."""
+    from app.services.auth import create_access_token
+
+    # Create token for admin user
+    token = create_access_token(data={"sub": admin_user.name})
+
+    # Get user from token
+    user = get_current_user(token=token, db=db_session)
+    assert user is not None
+    assert user.name == admin_user.name
+    assert user.id == admin_user.id
+
+    # Test invalid token
+    with pytest.raises(HTTPException) as exc_info:
+        get_current_user(token="invalid_token")
+    assert exc_info.value.status_code == status.HTTP_401_UNAUTHORIZED
+
+    # Test expired token (create token with past expiration)
+    past_expire = timedelta(minutes=-100)
+    expired_token = create_access_token(
+        data={"sub": admin_user.name}, expires_delta=past_expire
+    )
+
+    with pytest.raises(HTTPException) as exc_info:
+        get_current_user(token=expired_token)
+    assert exc_info.value.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_auth_is_admin(db_session, admin_user, regular_user):
+    """Test admin authorization check."""
+    from app.services.auth import create_access_token
+
+    # Create token for admin user
+    admin_token = create_access_token(data={"sub": admin_user.name})
+
+    # Admin should pass
+    result = auth_is_admin(token=admin_token, db=db_session)
+    assert result is True
+
+    # Create token for regular user
+    user_token = create_access_token(data={"sub": regular_user.name})
+
+    # Regular user should fail
+    with pytest.raises(HTTPException) as exc_info:
+        auth_is_admin(token=user_token, db=db_session)
+    assert exc_info.value.status_code == status.HTTP_403_FORBIDDEN
+
+
+def test_create_first_user(db_session):
+    """Test automatic creation of first admin user."""
+    # Clear any existing users
+    from sqlmodel import select
+
+    db_session.exec(select(UserDB)).all()
+    for user in db_session.exec(select(UserDB)).all():
+        db_session.delete(user)
+    db_session.commit()
+
+    # Create first user
+    result = create_first_user(db=db_session)
+    assert result is not None
+    assert result.name == "admin"
+    assert result.is_admin is True
+
+    # Verify user exists in database
+    user = db_session.exec(select(UserDB).where(UserDB.name == "admin")).first()
+    assert user is not None
+    assert user.is_admin is True
+
+    # Test that it doesn't create another admin if one exists
+    second_result = create_first_user(db=db_session)
+    assert second_result is None  # Should print "Admin user already exists"
+
+
+def test_token_endpoint(client, admin_user):
+    """Test the token endpoint for login."""
+    # Test successful login
+    response = client.post(
+        "/api/v1/token", data={"username": admin_user.name, "password": "admin123"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert "access_token" in data
+    assert data["token_type"] == "bearer"
+
+    # Test failed login with wrong password
+    response = client.post(
+        "/api/v1/token", data={"username": admin_user.name, "password": "wrongpassword"}
+    )
+    assert response.status_code == 401
+
+    # Test failed login with non-existent user
+    response = client.post(
+        "/api/v1/token", data={"username": "nonexistent", "password": "password"}
+    )
+    assert response.status_code == 401
+
+
+def test_test_login_endpoint(client, admin_user, auth_headers):
+    """Test the test login endpoint."""
+    # Test with valid token
+    response = client.get("/api/v1/test/login", headers=auth_headers)
+    assert response.status_code == 200
+    data = response.json()
+    assert data["name"] == admin_user.name
+    assert data["is_admin"] is True
+
+    # Test without token
+    response = client.get("/api/v1/test/login")
+    assert response.status_code == 401

test/test_services/test_caldav.py

@@ -0,0 +1,55 @@
+from datetime import UTC, date, datetime
+
+from app.services.caldav import CaldavEventCache
+
+
+def create_cache():
+    """Create the instance without calling __init__, so tests need no CalDAV server."""
+    cache = CaldavEventCache.__new__(CaldavEventCache)
+    cache.events = [
+        {
+            "uid": "1",
+            "summary": "Timed event",
+            "groups": ["team-a", "shared"],
+            "start": datetime(2026, 11, 2, 10, 0, tzinfo=UTC),
+            "end": datetime(2026, 11, 2, 11, 0, tzinfo=UTC),
+            "all_day": False,
+        },
+        {
+            "uid": "2",
+            "summary": "All-day event",
+            "groups": ["team-b", "shared"],
+            "start": date(2026, 11, 2),
+            "end": date(2026, 11, 3),  # Exclusive end date
+            "all_day": True,
+        },
+    ]
+    return cache
+
+
+def test_active_during_timed_event():
+    cache = create_cache()
+    during_event = datetime(2026, 11, 2, 10, 30, tzinfo=UTC)
+    assert cache.is_group_active("team-a", during_event)
+    assert cache.is_group_active("shared", during_event)
+    assert cache.is_group_active("team-b", during_event)  # warum 'not'?
+
+
+def test_timed_event_end_exclusive():
+    cache = create_cache()
+    at_timed_event_end = datetime(2026, 11, 2, 11, 0, tzinfo=UTC)
+    assert not cache.is_group_active("team-a", at_timed_event_end)
+    assert cache.is_group_active("team-b", at_timed_event_end)
+    assert cache.is_group_active("shared", at_timed_event_end)
+
+
+def test_unknown_group():
+    cache = create_cache()
+    during_event = datetime(2026, 11, 2, 10, 30, tzinfo=UTC)
+    assert not cache.is_group_active("unknown", during_event)
+
+
+def test_all_day_event_exclusive():
+    cache = create_cache()
+    next_day = datetime(2026, 11, 3, 9, 0, tzinfo=UTC)
+    assert not cache.is_group_active("team-b", next_day)

test/test_services/test_card_manager.py

@@ -0,0 +1,62 @@
+def test_get_cards_for_group(client, auth_headers, test_group, test_card):
+    """Test getting all cards for a group."""
+    response = client.get(f"/api/v1/cards/{test_group.id}", headers=auth_headers)
+    assert response.status_code == 200
+
+    cards = response.json()
+    assert len(cards) >= 1
+    assert any(card["id"] == test_card.id for card in cards)
+
+
+def test_get_cards_for_nonexistent_group(client, auth_headers):
+    """Test getting cards for a non-existent group."""
+    response = client.get("/api/v1/cards/99999", headers=auth_headers)
+    assert response.status_code == 200
+
+    cards = response.json()
+    assert len(cards) == 0  # Empty list for non-existent group
+
+
+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("/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  # noqa: E712
+    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("/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

@@ -0,0 +1,60 @@
+from alembic.config import Config
+from alembic.script import ScriptDirectory
+from sqlalchemy import create_engine, text
+from sqlmodel import Session
+
+from app.model.models import UserDB
+from app.services.database import (
+    add_and_refresh,
+    configure_engine,
+    create_db_and_tables,
+    get_engine,
+    reset_engine,
+)
+from app.services.settings import settings
+
+
+def test_configure_engine_and_reset(tmp_path, db_session):
+    """configure_engine() is the single injection point for the whole app.
+    reset_engine() drops the engine again and rebuilds from settings
+    conftest's autouse _installed_engine fixture puts the shared test engine back.
+    """
+    from app.services import database as database_service
+
+    url = f"sqlite:///{tmp_path / 'injected.db'}"
+    injected = create_engine(url, connect_args={"check_same_thread": False})
+    configure_engine(injected)
+    assert get_engine() is injected
+
+    # Clean database: create_db_and_tables() must create the file and stamp
+    # *the configured engine*, not the URL from alembic.ini/settings.
+    create_db_and_tables()
+    assert tmp_path.joinpath("injected.db").exists()
+    with Session(injected) as session:
+        stamped = session.exec(text("SELECT version_num FROM alembic_version")).all()
+    head = ScriptDirectory.from_config(
+        Config(settings.alembic_config)
+    ).get_current_head()
+    assert [row[0] for row in stamped] == [head]
+
+    reset_engine()
+    rebuilt = get_engine()
+    assert rebuilt is not injected
+    assert str(rebuilt.url) == str(database_service.settings.sqlalchemy_database_url)
+
+
+def test_add_and_refresh(db_session):
+    """Test add_and_refresh helper function."""
+    user = UserDB(name="Test User", passwordhash="hashed")
+
+    # Add user
+    result = add_and_refresh(db_session, user)
+
+    # Assert that user is now in database with ID
+    assert result.id is not None
+    assert result.name == "Test User"
+
+    # Verify in database
+    db_user = db_session.get(UserDB, result.id)
+    assert db_user is not None
+    assert db_user.name == "Test User"

test/test_services/test_door.py

@@ -0,0 +1,196 @@
+import datetime
+
+import pytest
+
+from app.model.models import (
+    AccessAuthorizationDB,
+    Card,
+    GroupDB,
+    OneShotAccess,
+)
+from app.services.door import checkAccess
+
+
+def test_check_access_with_valid_oneshot(db_session):
+    # Setup: create card with valid access
+    group = GroupDB(name="Test Group")
+    db_session.add(group)
+    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",
+    )
+    db_session.add(card)
+
+    oneshot = OneShotAccess(
+        uses=1, ends_at=datetime.datetime.now() + datetime.timedelta(days=1)
+    )
+    db_session.add(oneshot)
+
+    aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="oneshot")
+    db_session.add(aa)
+    aa.oneshot = oneshot
+    group.accessauths = [aa]
+
+    db_session.commit()
+
+    # Test: access should be granted within time window
+    assert checkAccess("test-key-123", db_session)
+    assert aa.oneshot.uses == 0
+
+
+def test_check_access_invalid_card(db_session):
+    # Should raise exception for non-existent card
+    with pytest.raises(Exception, match="No Access"):
+        checkAccess("non-existent-key", db_session)
+
+
+def test_check_access_with_inactive_aa(db_session):
+    # Setup: create card with valid access
+    group = GroupDB(name="Test Group")
+    db_session.add(group)
+    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",
+    )
+    db_session.add(card)
+
+    oneshot = OneShotAccess(
+        uses=1, ends_at=datetime.datetime.now() + datetime.timedelta(days=1)
+    )
+    db_session.add(oneshot)
+    aa = AccessAuthorizationDB(name="Test oneshot AA", is_active=False, type="oneshot")
+    db_session.add(aa)
+    aa.oneshot = oneshot
+    group.accessauths = [aa]
+
+    db_session.commit()
+
+    assert not checkAccess("test-key-123", db_session)
+
+
+def test_check_access_with_inactive_card(db_session):
+    # Setup: create card with valid access
+    group = GroupDB(name="Test Group")
+    db_session.add(group)
+    db_session.commit()
+
+    card = Card(
+        key="test-key-123",
+        group_id=group.id,
+        enabled=False,
+        name="test_card",
+        card_serial="00:00:00:00:00:00:00",
+    )
+    db_session.add(card)
+
+    oneshot = OneShotAccess(
+        uses=1, ends_at=datetime.datetime.now() + datetime.timedelta(days=1)
+    )
+    db_session.add(oneshot)
+    aa = AccessAuthorizationDB(name="Test oneshot AA", is_active=True, type="oneshot")
+    db_session.add(aa)
+    aa.oneshot = oneshot
+    group.accessauths = [aa]
+
+    db_session.commit()
+
+    assert not checkAccess("test-key-123", db_session)
+
+
+def test_regression_check_access_issue_28(db_session):
+    # Having a active aa after a inactive/invalid one denies access
+    group = GroupDB(name="Test Group")
+    db_session.add(group)
+    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",
+    )
+    db_session.add(card)
+
+    # Creating oneshot with inactive AA
+    oneshot1 = OneShotAccess(
+        uses=1, ends_at=datetime.datetime.now() + datetime.timedelta(days=1)
+    )
+    db_session.add(oneshot1)
+    aa1 = AccessAuthorizationDB(name="First AA", is_active=False, type="oneshot")
+    db_session.add(aa1)
+    aa1.oneshot = oneshot1
+
+    # Creating oneshot with active AA
+    oneshot2 = OneShotAccess(
+        uses=1, ends_at=datetime.datetime.now() + datetime.timedelta(days=1)
+    )
+    db_session.add(oneshot2)
+    aa2 = AccessAuthorizationDB(name="Second AA", is_active=True, type="oneshot")
+    db_session.add(aa2)
+    aa2.oneshot = oneshot2
+
+    group.accessauths = [aa1, aa2]
+
+    db_session.commit()
+
+    assert checkAccess("test-key-123", db_session)
+
+
+# Berlin's DST rules as a POSIX TZ string: +01:00 in winter, +02:00 in summer.
+BERLIN_TZ = "CET-1CEST,M3.5.0,M10.5.0/3"
+
+
+def _create_card_with_oneshot(db_session, ends_at):
+    """Create a card whose only authorization is a oneshot ending at ends_at."""
+    group = GroupDB(name="Test Group")
+    db_session.add(group)
+    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",
+    )
+    db_session.add(card)
+
+    oneshot = OneShotAccess(uses=1, ends_at=ends_at)
+    db_session.add(oneshot)
+
+    aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="oneshot")
+    db_session.add(aa)
+    aa.oneshot = oneshot
+    group.accessauths = [aa]
+
+    db_session.commit()
+
+
+@pytest.mark.parametrize("system_tz", [BERLIN_TZ], indirect=True)
+def test_check_access_oneshot_end_time_is_compared_in_utc(db_session, system_tz):
+    """A oneshot ending in 30 minutes is valid even though the local clock is
+    ahead of UTC: the stored end time is UTC and must not be read as local time."""
+    ends_at = datetime.datetime.now(datetime.UTC) + datetime.timedelta(minutes=30)
+    _create_card_with_oneshot(db_session, ends_at)
+
+    assert checkAccess("test-key-123", db_session)
+
+
+@pytest.mark.parametrize("system_tz", [BERLIN_TZ], indirect=True)
+def test_check_access_oneshot_expired(db_session, system_tz):
+    """A oneshot that already ended must be rejected whatever the local clock is."""
+    ends_at = datetime.datetime.now(datetime.UTC) - datetime.timedelta(minutes=30)
+    _create_card_with_oneshot(db_session, ends_at)
+
+    assert not checkAccess("test-key-123", db_session)

test/test_services/test_group_manager.py

@@ -0,0 +1,62 @@
+def test_create_group(client, auth_headers):
+    """Test creating a new group."""
+    group_data = {"name": "New Test Group"}
+
+    response = client.post("/api/v1/groups/", json=group_data, headers=auth_headers)
+    assert response.status_code == 200
+
+    data = response.json()
+    assert data["name"] == "New Test Group"
+    assert "id" in data
+
+
+def test_create_duplicate_group(client, auth_headers, test_group):
+    """Test creating a group with a duplicate name."""
+    group_data = {"name": test_group.name}
+
+    response = client.post("/api/v1/groups/", json=group_data, headers=auth_headers)
+    # This should fail due to unique constraint
+    assert response.status_code == 409  # Validation error
+
+
+def test_get_groups(client, auth_headers, test_group):
+    """Test retrieving all groups."""
+    response = client.get("/api/v1/groups/", headers=auth_headers)
+    assert response.status_code == 200
+
+    groups = response.json()
+    assert len(groups) >= 1
+
+    group_names = [group["name"] for group in groups]
+    assert test_group.name in group_names
+
+
+def test_delete_group(client, auth_headers, test_group):
+    """Test deleting a group."""
+    response = client.delete(f"/api/v1/groups/{test_group.id}", headers=auth_headers)
+    assert response.status_code == 200
+    assert "deleted successfully" in response.json()["message"].lower()
+
+    # Verify group is deleted
+    response = client.get("/api/v1/groups/", headers=auth_headers)
+    groups = response.json()
+    assert not any(group["id"] == test_group.id for group in groups)
+
+
+def test_delete_nonexistent_group(client, auth_headers):
+    """Test deleting a non-existent group."""
+    response = client.delete("/api/v1/groups/99999", headers=auth_headers)
+    assert response.status_code == 404
+
+
+def test_group_operations_by_non_admin(client, user_auth_headers):
+    """Test that non-admin users cannot perform group operations."""
+    # Try to create a group
+    response = client.post(
+        "/api/v1/groups/", json={"name": "test"}, headers=user_auth_headers
+    )
+    assert response.status_code == 403
+
+    # Try to get groups
+    response = client.get("/api/v1/groups/", headers=user_auth_headers)
+    assert response.status_code == 403

test/test_services/test_settings.py

@@ -0,0 +1,93 @@
+"""Checks that the suite never resolves settings from .env or the ambient OS env."""
+
+import json
+import os
+
+from app.services.settings import Settings, settings
+from test.conftest import TEST_ENV
+
+# Values a developer's .env / shell would plausibly contribute.
+HOSTILE_ENV = {
+    "MOCK": "False",
+    "DISABLE_CALDAV": "False",
+    "DISABLE_CARDS": "False",
+    "CALDAV_URL": "https://caldav.example.com/dav.php",
+    "CALDAV_USER": "real-user",
+    "CALDAV_PASSWORD": "real-password",
+    "CALDAV_CALENDARS": '["Family"]',
+    "CALDAV_REFRESH_INTERVAL": "1",
+    "MIFARE_APP_MASTER_KEY": "a" * 16,
+    "MIFARE_ACL_READ_BASE_KEY": "b" * 16,
+    "MIFARE_ACL_WRITE_BASE_KEY": "c" * 16,
+    "SQLALCHEMY_DATABASE_URL": "postgresql://db.example.com:5432/gatekeeper",
+    "MQTT_HOST": "mqtt.example.com",
+    "SECRET_KEY": "leaked-secret",
+}
+
+
+def test_every_settings_field_is_pinned_by_the_test_env():
+    ambient = {name.upper() for name in Settings.model_fields}
+
+    missing = sorted(ambient - set(TEST_ENV))
+    assert not missing, (
+        f"Settings fields not pinned by conftest.TEST_ENV: {missing} — their "
+        "value could come from .env or the OS environment"
+    )
+
+    stale = [name for name in TEST_ENV if name not in ambient]
+    assert not stale, f"TEST_ENV names that are no longer Settings fields: {stale}"
+
+
+def test_ambient_env_and_dotenv_could_not_supply_any_value():
+    """No field is left to .env: each one has a pinned value in os.environ."""
+    ambient = {name.upper() for name in Settings.model_fields}
+    unpinned = sorted(ambient - set(TEST_ENV))
+    assert not unpinned, f"unpinned Settings fields: {unpinned}"
+
+    for name, expected in TEST_ENV.items():
+        assert os.environ.get(name) == expected, (
+            f"conftest did not pin {name} (os.environ={os.environ.get(name)!r})"
+        )
+
+    for name, hostile in HOSTILE_ENV.items():
+        leaked = os.environ.get(name)
+        assert leaked != hostile, f"{name} leaked from the ambient env ({leaked!r})"
+
+
+def test_shared_settings_come_from_the_test_env():
+    for name, expected in TEST_ENV.items():
+        actual = getattr(settings, name.lower())
+        assert _matches(actual, expected), (
+            f"{name}: settings={actual!r} != test value {expected!r}"
+        )
+
+
+def _matches(actual, expected: str) -> bool:
+    """Compare a pydantic-parsed value against the raw string pinned in TEST_ENV."""
+    if str(actual) == expected:
+        return True
+    try:
+        return actual == json.loads(expected)
+    except json.JSONDecodeError:
+        return False
+
+
+def test_env_vars_outrank_env_file(monkeypatch):
+    """Why the scrub matters: an injected env value wins over TEST_ENV/.env.
+
+    Scoped to this test (monkeypatch restores os.environ on teardown), which is how
+    a test that needs other settings must do it.
+    """
+    for name, value in HOSTILE_ENV.items():
+        monkeypatch.setenv(name, value)
+
+    leaked = Settings()
+    assert leaked.mock is False
+    assert leaked.disable_caldav is False
+    assert leaked.sqlalchemy_database_url.startswith("postgresql://")
+    assert leaked.mqtt_host == "mqtt.example.com"
+
+    # The shared settings object the rest of the suite uses is untouched.
+    assert settings.mock is True
+    assert settings.disable_caldav is True
+    assert settings.disable_cards is True

test/test_services/test_user_manager.py

@@ -0,0 +1,150 @@
+def test_create_user(client, auth_headers):
+    """Test creating a new user."""
+    user_data = {
+        "name": "newuser",
+        "email": "newuser@example.com",
+        "is_admin": False,
+        "password": "newpassword123",
+    }
+
+    response = client.post("/api/v1/users/", json=user_data, headers=auth_headers)
+    assert response.status_code == 200
+
+    data = response.json()
+    assert data["name"] == "newuser"
+    assert data["email"] == "newuser@example.com"
+    assert data["is_admin"] is False
+    assert "id" in data
+    assert "passwordhash" not in data  # Password hash should not be in response
+
+
+def test_create_user_unauthorized(client):
+    """Test creating a user without admin credentials."""
+    user_data = {
+        "name": "unauthorized_user",
+        "email": "unauthorized@example.com",
+        "password": "password123",
+    }
+
+    response = client.post("/api/v1/users/", json=user_data)
+    assert response.status_code == 401
+
+
+def test_get_users(client, auth_headers, admin_user, regular_user):
+    """Test retrieving all users."""
+    response = client.get("/api/v1/users/", headers=auth_headers)
+    assert response.status_code == 200
+
+    users = response.json()
+    assert len(users) >= 2  # At least admin_user and regular_user
+
+    user_names = [user["name"] for user in users]
+    assert admin_user.name in user_names
+    assert regular_user.name in user_names
+
+
+def test_get_user_by_id(client, auth_headers, regular_user):
+    """Test retrieving a specific user by ID."""
+    response = client.get(f"/api/v1/users/{regular_user.id}", headers=auth_headers)
+    assert response.status_code == 200
+
+    data = response.json()
+    assert data["id"] == regular_user.id
+    assert data["name"] == regular_user.name
+
+
+def test_get_current_user(client, auth_headers, admin_user):
+    """Test getting the special url current"""
+    response = client.get("/api/v1/users/current", headers=auth_headers)
+    assert response.status_code == 200
+
+    data = response.json()
+    assert data["id"] == admin_user.id
+    assert data["name"] == admin_user.name
+    assert data["is_admin"] == admin_user.is_admin
+
+
+def test_get_nonexistent_user(client, auth_headers):
+    """Test retrieving a non-existent user."""
+    response = client.get("/api/v1/users/99999", headers=auth_headers)
+    assert response.status_code == 404
+    assert "not found" in response.json()["detail"].lower()
+
+
+def test_update_user(client, auth_headers, regular_user):
+    """Test updating a user."""
+    update_data = {"name": "updated_name", "email": "updated@example.com"}
+
+    response = client.patch(
+        f"/api/v1/users/{regular_user.id}", json=update_data, headers=auth_headers
+    )
+    assert response.status_code == 200
+
+    data = response.json()
+    assert data["name"] == "updated_name"
+    assert data["email"] == "updated@example.com"
+    # Unchanged fields should remain the same
+    assert data["is_admin"] == regular_user.is_admin
+
+
+def test_update_user_password(client, auth_headers, regular_user):
+    """Test updating a user's password."""
+    update_data = {"password": "new_password_456"}
+
+    response = client.patch(
+        f"/api/v1/users/{regular_user.id}", json=update_data, headers=auth_headers
+    )
+    assert response.status_code == 200
+
+    # Verify password can be used for login
+    login_response = client.post(
+        "/api/v1/token",
+        data={"username": regular_user.name, "password": "new_password_456"},
+    )
+    assert login_response.status_code == 200
+
+
+def test_update_nonexistent_user(client, auth_headers):
+    """Test updating a non-existent user."""
+    update_data = {"name": "updated"}
+    response = client.patch(
+        "/api/v1/users/99999", json=update_data, headers=auth_headers
+    )
+    assert response.status_code == 404
+
+
+def test_delete_user(client, auth_headers, regular_user):
+    """Test deleting a user."""
+    response = client.delete(f"/api/v1/users/{regular_user.id}", headers=auth_headers)
+    assert response.status_code == 200
+    assert "deleted successfully" in response.json()["message"].lower()
+
+    # Verify user is deleted
+    response = client.get(f"/api/v1/users/{regular_user.id}", headers=auth_headers)
+    assert response.status_code == 404
+
+
+def test_delete_nonexistent_user(client, auth_headers):
+    """Test deleting a non-existent user."""
+    response = client.delete("/api/v1/users/99999", headers=auth_headers)
+    assert response.status_code == 404
+
+
+def test_user_operations_by_non_admin(client, user_auth_headers):
+    """Test that non-admin users cannot perform admin operations."""
+    # Try to create a user
+    response = client.post(
+        "/api/v1/users/",
+        json={"name": "test", "password": "pass"},
+        headers=user_auth_headers,
+    )
+    assert response.status_code == 403
+
+    # Try to get users
+    response = client.get("/api/v1/users/", headers=user_auth_headers)
+    assert response.status_code == 403
+
+    # Try to delete the admin user (if ID is known)
+    # This would require knowing the admin user ID
+    # response = client.delete(f"/api/v1/users/{admin_id}", headers=user_auth_headers)
+    # assert response.status_code == 403