36 Commits
Author SHA1 Message Date
ahtlon 6b98e65a41 Fix #27. Also clear up error messages a bit 2026-08-08 00:26:02 +02:00
ahtlon c5934d05c6 Fix #28 2026-08-07 23:41:13 +02:00
ahtlon 0bcc83871d [Door] switch gpio out to mqtt 2026-08-05 03:22:55 +02:00
ahtlon 4b797997d4 Switch lgpio for mqtt 2026-08-05 02:24:05 +02:00
ahtlon f635149d1f Check for disabled card and add tests 2026-08-05 00:52:46 +02:00
ahtlon bbb41e41ab Add active check 2026-08-05 00:27:08 +02:00
ahtlon 0497d1bbcf seperate dev/normal deps 2026-08-05 00:27:08 +02:00
ahtlon c5c428c621 Merge pull request 'door section' (#21) from dev into master
Reviewed-on: #21
2026-08-03 00:39:03 +02:00
ahtlon 023a704e36 Make scanning the keycard while open close the door 2026-08-01 23:27:54 +02:00
ahtlon 73ff546c70 Update readme 2026-08-01 14:47:14 +02:00
ahtlon 55072bcc1d Fix module 2026-08-01 14:47:08 +02:00
ahtlon c9f740b7a3 Fix lgpio build... 2026-08-01 14:46:17 +02:00
ahtlon 936b786659 Switch gpiozero(fuck you) for lgpio 2026-08-01 14:45:44 +02:00
ahtlon 158b430235 [settings] fuck it, rework the settings again
this time using pydantic-settings as a base
- removed all os.getenv calls
- removed the secret_key default option
- reworked database loading, creating tables
- prob. something else also but its 4:30 and i have to sleep
2026-07-31 04:33:18 +02:00
ahtlon bad4e8dd1b [settings] rework settings 2026-07-31 01:50:08 +02:00
ahtlon b7b07b82da [door] correct spelling, add status endpoint
the door section of the api should "just work" now
2026-07-31 00:38:40 +02:00
ahtlon 717ce951a6 [door] add doorcontroller with mock pins for testing 2026-07-31 00:36:11 +02:00
ahtlon d37526e39a [pyproject] add gpiozero 2026-07-31 00:33:13 +02:00
ahtlon 40312020c8 Merge pull request 'merge dev to master for infra pr' (#20) from dev into master
Reviewed-on: #20
2026-07-30 18:35:58 +02:00
ahtlon 04a76f91fe [flake] the module now actually starts the prod server 2026-07-30 18:34:21 +02:00
ahtlon a8e80ed19d [main] fix startup when there's no db 2026-07-29 18:37:34 +02:00
ahtlon 7a70b6b1d4 [lint] more linting changes 2026-07-29 03:30:54 +02:00
ahtlon 3f20cdeed9 just_found_out_about_linters (#17)
This changed a bunch of code but no

Reviewed-on: #17
Co-authored-by: ahtlon <git@ahtlon.de>
Co-committed-by: ahtlon <git@ahtlon.de>
2026-07-29 02:46:50 +02:00
ahtlon 683cd9a545 [main.py] move settings stuff into its own import
also correct imports to use app namespace
2026-07-29 01:24:22 +02:00
ahtlon 1dfd01fc71 [CORS] fix methods
maybe add test for cors? idk how tho
2026-07-29 01:22:29 +02:00
ahtlon 54c7fe1ed2 [alembic] automaticly stamp new database with the "head" tag so migrations are easier.
also fix nix develop due to multiple top level folders
2026-07-28 16:14:33 +02:00
ahtlon 48d272d4d6 [alembic] generate base 2026-07-28 00:17:35 +02:00
ahtlon 7e27026da9 [uv] Add alembic 2026-07-27 23:58:20 +02:00
ahtlon 29c013a121 [nix] add module.nix 2026-07-27 23:50:22 +02:00
ahtlon 5f1030c831 Merge pull request 'Change accessauth, timetable model' (#16) from change_tt_model into master
Reviewed-on: #16
Fixes #9
2026-07-27 17:19:58 +02:00
ahtlon ea2a1f916d [Tests] add new tests for oneshot type aa 2026-07-27 17:19:10 +02:00
ahtlon 1efbad1db3 [Tests] Fix tests for new model 2026-07-27 17:19:10 +02:00
ahtlon 4b87603e06 [AA] fix patching AA with oneshots 2026-07-27 17:19:10 +02:00
ahtlon 5f19409da4 [AA] Add oneshot to checkAccess function 2026-07-27 17:19:10 +02:00
ahtlon e45abb5c55 [AA] fix the db insert 2026-07-27 17:19:10 +02:00
ahtlon 47b84a097c [Models] This doesn't quite work yet but this is about the model im thinking of for the future 2026-07-27 17:19:10 +02:00
37 changed files with 1761 additions and 639 deletions
+4
View File
@@ -1,4 +1,8 @@
{
"[python]": {
"editor.formatOnSave": true,
"editor.defaultFormatter": "charliermarsh.ruff"
},
"python.testing.pytestArgs": [
"test"
],
+3 -6
View File
@@ -1,9 +1,10 @@
## Gatekeeper - Door access system
#### Status: WIP - getting there o.o
#### Status: "WORKING" - Base functionality is there, mayor issues
Start prod server `nix run`<br>
Start dev server `nix run .#dev` or `nix run .#dev -- {args}`<br>
Interactive dev with `nix develop`, then sync deps with `uv sync`<br>
There is a nix module you can use by importing `inputs.gatekeeper.nixosModules.gatekeeper`<br>
Swagger UI @ http://127.0.0.1:8000/api/v1/docs<br>
OpenApi @ http://127.0.0.1:8000/api/v1/openapi.json<br>
@@ -33,17 +34,13 @@ Range: The range of the ACR1552U was much better at over 60mm (almost 70mm if yo
#### Issues:
- cards can only unlock, not lock
- documentation missing
- raspberry pi image not working
- no door state
- no door operations
- hardcoded secret key in auth.py -> centralise env var loading
- i don't like the error handling in the scanner - doesn't pass errors correctly
- cors for frontend: https://fastapi.tiangolo.com/tutorial/cors
- Load cors from env var or something
- BackgroundScanner shouldn't get a single session for the whole lifecycle
- input validation maybe
- too many imports
- inconsistent logging (request logging?)
- rate limiting maybe
- pretty sure the controllers are doing too much stuff
+149
View File
@@ -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
+1
View File
@@ -0,0 +1 @@
Generic single-database configuration.
+80
View File
@@ -0,0 +1,80 @@
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
# 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
# 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 = config.get_main_option("sqlalchemy.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(
config.get_section(config.config_ini_section, {}),
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()
+28
View File
@@ -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"}
+28
View File
@@ -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
+3 -1
View File
@@ -6,4 +6,6 @@ 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
#THESE ARE TESTING KEYS - DO NOT USE IN PROD
SQLALCHEMY_DATABASE_PATH="./gatekeeper.db"
-6
View File
@@ -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)
+92 -25
View File
@@ -1,44 +1,85 @@
import logging
logger = logging.getLogger(__name__)
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 ..services.database import engine, get_session, add_and_refresh
from ..services.auth import auth_is_admin
import uuid as gen_uuid
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import selectinload
from sqlmodel import Session, select
from app.model.models import (
AccessAuthorizationCreate,
AccessAuthorizationDB,
AccessAuthorizationResponse,
AccessAuthorizationUpdate,
GroupDB,
GroupResponse,
OneShotAccess,
Timetable,
)
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.post("/", response_model=AccessAuthorizationResponse)
def add_accessauth(*, db: Session = Depends(get_session), aa: AccessAuthorizationCreate, admin: bool = Depends(auth_is_admin)):
def add_accessauth(
*,
db: Session = Depends(get_session),
aa: AccessAuthorizationCreate,
admin: bool = Depends(auth_is_admin),
):
logger.info(f"Creating accessauth with data: {aa}")
timetables = [Timetable.model_validate(t) for t in aa.timetables]
if aa.timetables != []:
timetables = [Timetable.model_validate(t) for t in aa.timetables]
else:
timetables = []
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
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), admin: bool = Depends(auth_is_admin)):
@aa_router.get("/", response_model=list[AccessAuthorizationResponse])
def get_all_accessauths(
db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
):
return db.exec(
select(AccessAuthorizationDB)
.options(selectinload(AccessAuthorizationDB.timetables))
).all()
select(AccessAuthorizationDB).options(
selectinload(AccessAuthorizationDB.timetables)
)
).all()
@aa_router.get("/{aa_id}", response_model=AccessAuthorizationResponse)
def get_one_accessauth(*, db: Session = Depends(get_session), aa_id: int, admin: bool = Depends(auth_is_admin)):
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, admin: bool = Depends(auth_is_admin)):
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")
@@ -46,12 +87,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=status.HTTP_409_CONFLICT, 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, admin: bool = Depends(auth_is_admin)):
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")
@@ -63,25 +113,42 @@ def unassign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa
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, admin: bool = Depends(auth_is_admin)):
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.model_dump(exclude_unset=True)
aa_data = aa.model_dump(exclude_unset=True, exclude_none=True)
if "timetables" in aa_data and aa_data["timetables"] is not None:
db_aa.timetables.clear()
timetables = [Timetable.model_validate(t) for t in aa_data["timetables"]]
db_aa.timetables = timetables
aa_data.pop("timetables")
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, admin: bool = Depends(auth_is_admin)):
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"}
+66 -24
View File
@@ -1,70 +1,112 @@
import logging
logger = logging.getLogger(__name__)
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select
from typing import List
from sqlalchemy.exc import NoResultFound
from ..model.models import Card, CardCreate, CardUpdate, GroupDB
from ..services.database import engine, get_session, add_and_refresh
from ..services.auth import auth_is_admin
import uuid as gen_uuid
from app.services.scanner import WriteNewCard, DeleteCard
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.exc import NoResultFound
from sqlmodel import Session, select
from app.model.models import Card, CardCreate, CardUpdate, GroupDB
from app.services.auth import auth_is_admin
from app.services.database import add_and_refresh, get_session
from app.services.scanner import DeleteCard, WriteNewCard
logger = logging.getLogger(__name__)
card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"])
def register_card(cardInput: CardCreate):
key, uid = WriteNewCard()
if key == None:
if key is None:
logger.info("No card registered. Check logs!")
raise HTTPException(status.HTTP_417_EXPECTATION_FAILED, detail="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
)
enabled=cardInput.enabled,
)
return card
@card_router.post("/", response_model=Card)
def add_card(cardInput: CardCreate, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
try: assert db.exec(select(Card).where(Card.name == cardInput.name)).one_or_none() == None
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:
raise HTTPException(status.HTTP_409_CONFLICT, detail="Name already used!")
try: assert db.exec(select(GroupDB).where(GroupDB.id == cardInput.group_id)).one_or_none() is not None
try:
assert (
db.exec(
select(GroupDB).where(GroupDB.id == cardInput.group_id)
).one_or_none()
is not None
)
except AssertionError:
raise HTTPException(status.HTTP_404_NOT_FOUND, detail="GroupID not found!")
card = register_card(cardInput)
return add_and_refresh(db, card)
@card_router.delete("/")
def del_card(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
def del_card(
*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
):
key = DeleteCard()
logger.info(key)
try:
card = db.exec(select(Card).where(Card.key == key)).one()
except NoResultFound:
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. Please tell an admin about this. KEY={key}")
raise HTTPException(
status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Key on card not found in DB. Please tell an admin about this. KEY={key}", # noqa: E501
)
db.delete(card)
db.commit()
return {"message": "Card deleted successfully"}
@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)):
@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
@card_router.patch("/{card_id}", response_model=Card)
def update_card(card_id: int, cardInput: CardUpdate, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
def update_card(
card_id: int,
cardInput: CardUpdate,
db: Session = Depends(get_session),
admin: bool = Depends(auth_is_admin),
):
db_card = db.get(Card, card_id)
if db_card is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Card not found!")
card_data = cardInput.model_dump(exclude_unset=True, exclude_none=True)
if "group_id" in card_data:
try: assert db.exec(select(GroupDB).where(GroupDB.id == cardInput.group_id)).one_or_none() is not None
try:
assert (
db.exec(
select(GroupDB).where(GroupDB.id == cardInput.group_id)
).one_or_none()
is not None
)
except AssertionError:
raise HTTPException(status.HTTP_404_NOT_FOUND, detail="GroupID not found!")
db_card.sqlmodel_update(card_data)
return add_and_refresh(db, db_card)
return add_and_refresh(db, db_card)
+25 -12
View File
@@ -1,11 +1,14 @@
import logging
logger = logging.getLogger(__name__)
from fastapi import APIRouter, Depends, HTTPException
from app.services.auth import auth_is_admin
from sqlalchemy import exc
from sqlmodel import Session, select
import sqlalchemy.exc as exc
from app.model.models import *
from app.services.database import get_session, add_and_refresh
from app.model.models import Card
from app.services.auth import auth_is_admin
from app.services.database import add_and_refresh, get_session
logger = logging.getLogger(__name__)
debug_router = APIRouter(
prefix="/api/v1/debug",
@@ -13,20 +16,30 @@ debug_router = APIRouter(
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)):
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}")
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"
)
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:
@@ -37,14 +50,14 @@ def remove_card_manually(card_id: str, db: Session = Depends(get_session)):
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(f"Debug Setting: Getting cards.")
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
return out
+20 -8
View File
@@ -1,20 +1,32 @@
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends
from sqlmodel import Session
from app.services.database import get_session
from app.services.auth import auth_is_admin
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 = APIRouter(prefix="/api/v1/door",tags=["Door"])
@door_router.put("/open")
def open_door(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
doorService.opendoor()
doorService.openDoor()
@door_router.put("/close")
def open_door(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
doorService.closedoor()
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)
return doorService.checkAccess(input, db=db)
+26 -10
View File
@@ -1,31 +1,47 @@
from fastapi import APIRouter, HTTPException, Depends, status
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select
from typing import List
from ..model.models import GroupDB, GroupResponse, GroupCreate
from ..services.database import engine, get_session, add_and_refresh
from ..model.models import GroupCreate, GroupDB, GroupResponse
from ..services.auth import auth_is_admin
from ..services.database import add_and_refresh, get_session
group_router = APIRouter(prefix="/api/v1/groups", tags=["Group"])
@group_router.get("/", response_model=List[GroupResponse])
def get_groups(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
@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, admin: bool = Depends(auth_is_admin)):
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!")
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, admin: bool = Depends(auth_is_admin)):
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"}
+52 -16
View File
@@ -1,42 +1,73 @@
import logging
logger = logging.getLogger(__name__)
from fastapi import APIRouter, HTTPException, Depends, status
from sqlmodel import Session, select
from typing import List
from ..model.models import UserResponse, UserCreate, UserDB, UserUpdate
from ..services.database import engine, get_session, add_and_refresh
from ..services.auth import get_password_hash, get_current_user as auth_user, auth_is_admin
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select
from app.model.models import UserCreate, UserDB, UserResponse, UserUpdate
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
logger = logging.getLogger(__name__)
user_router = APIRouter(tags=["Users"], prefix="/api/v1/users")
@user_router.post("/", response_model=UserResponse)
def create_user(*, db: Session = Depends(get_session), user: UserCreate, admin: bool = Depends(auth_is_admin)):
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() == None
try:
assert (
db.exec(select(UserDB).where(UserDB.name == user.name)).one_or_none()
is None
)
except AssertionError:
raise HTTPException(status.HTTP_409_CONFLICT, detail="Name already used!")
db_user = UserDB.model_validate(user, update=hashed_password)
return add_and_refresh(db, db_user)
@user_router.get("/", response_model=List[UserResponse])
def read_users(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
@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("/current", response_model=UserResponse)
def get_current_user(db: Session = Depends(get_session), user: UserDB = Depends(auth_user)):
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)):
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("/{user_id}", response_model=UserResponse)
def update_user(*, db: Session = Depends(get_session), user_id: int, user: UserUpdate, admin: bool = Depends(auth_is_admin)):
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")
@@ -48,12 +79,17 @@ def update_user(*, db: Session = Depends(get_session), user_id: int, user: UserU
db_user.sqlmodel_update(user_data, update=hashed_password)
return add_and_refresh(db, db_user)
@user_router.delete("/{user_id}")
def delete_user(*, db: Session = Depends(get_session), user_id: int, admin: bool = Depends(auth_is_admin)):
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"}
+34 -28
View File
@@ -1,49 +1,55 @@
import logging
logger = logging.getLogger(__name__)
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import OAuth2PasswordBearer
from contextlib import asynccontextmanager
from dotenv import load_dotenv
from .controllers import userManager, cardManager, groupManager, aaManager, doorManager, debugManager
from .services.database import create_db_and_tables, get_db_session
from .services.auth import token_router, create_first_user
from app.controllers import (
aaManager,
cardManager,
debugManager,
doorManager,
groupManager,
userManager,
)
from app.services.auth import create_first_user, token_router
from app.services.database import create_db_and_tables, get_db_session
from app.services.door import DoorController, init_controller
from app.services.scanner import BackgroundScanner
from app.services.settings import settings
logger = logging.getLogger(__name__)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
scanner = BackgroundScanner(db=get_db_session())
logging.basicConfig(level=logging.INFO)
def checkDeps():
load_dotenv()
MIFARE_APP_MASTER_KEY = os.getenv('MIFARE_APP_MASTER_KEY')
if not MIFARE_APP_MASTER_KEY:
logger.critical(f"MIFARE APP MASTER KEY not found!")
logger.critical("Writing and reading cards is disabled!")
@asynccontextmanager
async def lifespan(app: FastAPI):
logger.critical("-"*63)
logger.critical("---- Documentation is at http://127.0.0.1:8000/api/v1/docs ----")
logger.critical("-"*63)
checkDeps()
create_db_and_tables()
create_first_user(db=get_db_session())
logger.info("Database created and tables initialized.")
disableCards = os.getenv("DISABLE_CARDS")
if not disableCards:
init_controller(
DoorController(
mqtt_host=settings.mqtt_host,
mqtt_port=settings.mqtt_port,
mock_factory=settings.mock,
),
)
if not settings.disable_cards:
scanner.start()
logger.info("-" * 63)
logger.info("---- Documentation is at http://127.0.0.1:8000/api/v1/docs ----")
logger.info("-" * 63)
yield
#scanner.stop()
# scanner.stop()
app = FastAPI(
lifespan=lifespan,
docs_url="/api/v1/docs",
openapi_url="/api/v1/openapi.json"
)
lifespan=lifespan, docs_url="/api/v1/docs", openapi_url="/api/v1/openapi.json"
)
origins = [
"http://127.0.0.1",
@@ -55,7 +61,7 @@ app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["GET" "PUT" "POST" "DELETE" "PATCH"],
allow_methods=["GET", "PUT", "POST", "DELETE", "PATCH"],
allow_headers=["*"],
)
@@ -65,4 +71,4 @@ app.include_router(groupManager.group_router)
app.include_router(cardManager.card_router)
app.include_router(aaManager.aa_router)
app.include_router(doorManager.door_router)
app.include_router(debugManager.debug_router)
app.include_router(debugManager.debug_router)
+89 -16
View File
@@ -1,85 +1,153 @@
from sqlmodel import Field, Relationship, Session, SQLModel
from typing import List
from datetime import time
from datetime import datetime, time
from typing import Literal
from pydantic import model_validator
from sqlmodel import Field, Relationship, SQLModel
class Base(SQLModel):
pass
#### User
class UserBase(Base):
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)
passwordhash: str
class UserUpdate(Base):
name: str | None = None
email: str | None = None
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)
accessauth_id: int | None = Field(default=None, foreign_key="accessauthorizationdb.id", primary_key=True)
group_id: int | None = Field(
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")
accessauths: List["AccessAuthorizationDB"] = Relationship(back_populates="groups", link_model=AaGroupLink)
cards: list["Card"] = Relationship(back_populates="group")
accessauths: list["AccessAuthorizationDB"] = Relationship(
back_populates="groups", link_model=AaGroupLink
)
class GroupResponse(GroupBase):
id: int
cards: List["Card"] | None
accessauths: List["AccessAuthorizationDB"] | None
cards: list["Card"] | None
accessauths: list["AccessAuthorizationDB"] | None
#### AccessAuthorization
class AccessAuthorizationBase(Base):
name: str = Field(index=True)
type: Literal["timetable", "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)
timetables: List["Timetable"] = Relationship(back_populates="accessauth", cascade_delete=True)
type: str
groups: list["GroupDB"] = Relationship(
back_populates="accessauths", link_model=AaGroupLink
)
timetables: list["Timetable"] = Relationship(
back_populates="accessauth", cascade_delete=True
)
oneshot: "OneShotAccess" = Relationship(
back_populates="accessauth", cascade_delete=True
)
class OneShotAccessBase(Base):
uses: int = 1
ends_at: datetime
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"]
timetables: list["TimetableCreate"] = []
oneshot: OneShotAccessBase | None = None
@model_validator(mode="after")
def check_type(self):
if self.type == "timetable":
if not self.timetables:
raise ValueError(
"timetable auths require at least one timetable object"
)
if self.oneshot is not None:
raise ValueError("timetable auths are not allowed oneshot objects")
elif self.type == "oneshot":
if not self.oneshot:
raise ValueError("oneshot auths require a oneshot object")
if self.timetables:
raise ValueError("oneshot auths are not allowed timetable objects")
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"]
groups: List["GroupDB"]
timetables: list["Timetable"] = []
oneshot: OneShotAccessBase | None = None
groups: list["GroupDB"]
class AccessAuthorizationUpdate(Base):
name: str | None = None
type: Literal["timetable", "oneshot", "somefuturespec"] | None = None
is_active: bool | None = None
timetables: List["TimetableCreate"] | None = None
timetables: list["TimetableCreate"] | None = None
oneshot: OneShotAccessBase | None = None
#### Card
class Card(Base, table=True):
@@ -91,25 +159,30 @@ class Card(Base, table=True):
group_id: int | None = Field(default=None, foreign_key="groupdb.id")
group: GroupDB | None = Relationship(back_populates="cards")
class CardCreate(Base):
name: str = Field(unique=True, max_length=32)
enabled: bool = True
group_id: int
class CardUpdate(Base):
name: str | None = None
enabled: bool | None = None
group_id: int | None = None
class TimetableBase(Base):
weekday: int = Field(le=6, ge=0)
starttime: time
duration: int = Field(gt=0, lt=1440)
class Timetable(TimetableBase, 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="timetables")
class TimetableCreate(TimetableBase):
pass
+35 -28
View File
@@ -1,21 +1,23 @@
import logging
logger = logging.getLogger(__name__)
import secrets
import string
from datetime import UTC, datetime, timedelta
from typing import Annotated
from datetime import datetime, timedelta, timezone
from fastapi import APIRouter, HTTPException, Depends, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlmodel import Session, select
from pwdlib import PasswordHash
import jwt
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from ..model.models import UserDB, Token, TokenData, UserCreate
from ..services.database import *
import secrets, string, os
from pwdlib import PasswordHash
from sqlmodel import Session, select
from dotenv import load_dotenv
from app.model.models import Token, TokenData, UserDB
from app.services.database import add_and_refresh, get_session
from app.services.settings import settings
load_dotenv()
SECRET_KEY = os.getenv("SECRET_KEY", default="ff"*16)
logger = logging.getLogger(__name__)
SECRET_KEY = settings.secret_key
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 120
@@ -25,16 +27,20 @@ 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:
@@ -43,24 +49,26 @@ def authenticate_user(db, username: str, password: str):
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(timezone.utc) + expires_delta
expire = datetime.now(UTC) + expires_delta
else:
expire = datetime.now(timezone.utc) + timedelta(minutes=15)
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"}
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
@@ -75,29 +83,29 @@ def get_current_user(
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"}
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))
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
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}")
@@ -106,14 +114,14 @@ def create_first_user(db: Session):
@token_router.post("/token")
def login_for_access_token(
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
db: Session = Depends(get_session)
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"}
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
@@ -121,8 +129,7 @@ def login_for_access_token(
)
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
def test_login(current_user: Annotated[UserDB, Depends(get_current_user)]) -> UserDB:
return current_user
+37 -8
View File
@@ -1,23 +1,52 @@
from sqlmodel import create_engine, SQLModel, Session
import logging
from functools import lru_cache
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__)
@lru_cache
def get_engine():
return create_engine(
settings.sqlalchemy_database_url, connect_args={"check_same_thread": False}
)
engine = create_engine(SQLALCHEMY_DATABASE_URL)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
inspector = inspect(get_engine())
existing_tables = inspector.get_table_names()
if not existing_tables:
SQLModel.metadata.create_all(get_engine())
from alembic.config import Config
from alembic import command
alembic_cfg = Config(settings.alembic_config)
alembic_cfg.set_main_option("sqlalchemy.url", str(get_engine().url))
alembic_cfg.attributes["sqlalchemy.url"] = settings.sqlalchemy_database_url
command.stamp(alembic_cfg, "head")
logger.info("Database created and tables initialized.")
else:
logger.info(
"Database already initialized (%d tables found).", len(existing_tables)
)
def get_session():
with Session(engine) as db:
with Session(get_engine()) as db:
yield db
def get_db_session():
return Session(engine)
return Session(get_engine())
def add_and_refresh(db: Session, obj):
db.add(obj)
db.commit()
db.refresh(obj)
return obj
return obj
+116 -32
View File
@@ -1,52 +1,136 @@
import logging
logger = logging.getLogger(__name__)
from datetime import date, datetime, timedelta
from time import sleep
from sqlalchemy import exc
from sqlmodel import select
from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import selectinload
import sqlalchemy.exc as exc
import datetime
from app.services.database import Session, get_session
from app.model.models import *
from app.model.models import Card, OneShotAccess
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
doorIsOpen = True
# I think this could also be gpio controlled
#See: https://github.com/technyon/nuki_hub#gpio-lock-control-optional
def openDoor():
global doorIsOpen
doorIsOpen = True
logger.info("Still needs gpio out")
pass
get_controller().open()
def closeDoor():
global doorIsOpen
doorIsOpen = False
logger.info("Still needs gpio out")
pass
get_controller().close()
def isDoorOpen():
return doorIsOpen
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:
current_weekday = datetime.datetime.weekday(datetime.date.today())
current_time = datetime.datetime.now()
current_weekday = datetime.weekday(date.today())
current_time = datetime.now()
card = db.exec(select(Card).where(Card.key == key)).one()
if not card.enabled:
logger.info("Card Inactive!")
return False
for auth in card.group.accessauths:
logger.info(f"checking auth: {auth.name}")
for timetable in auth.timetables:
logger.info(f" checking timetable {timetable.id}")
logger.info(f" comparing weekday: CUR:{current_weekday} TT:{timetable.weekday}")
if current_weekday == timetable.weekday:
starttime = datetime.datetime.combine(datetime.date.today(), timetable.starttime)
endtime = starttime + datetime.timedelta(minutes=timetable.duration)
logger.info(f" comparing time: Start:{starttime} Current:{current_time} End:{endtime}")
if starttime < current_time < endtime:
logger.info("Access Valid!")
if not auth.is_active:
logger.info("AA inactive!")
continue
if auth.type == "timetable":
for timetable in auth.timetables:
logger.info(f" checking timetable {timetable.id}")
logger.info(
f" comparing weekday: CUR:{current_weekday} TT:{timetable.weekday}"
)
if current_weekday == timetable.weekday:
starttime = datetime.combine(date.today(), timetable.starttime)
endtime = starttime + timedelta(minutes=timetable.duration)
logger.info(
f" comparing time: Start:{starttime} Current:{current_time} End:{endtime}"
)
if starttime < current_time < endtime:
logger.info("Access Valid!")
return True
if auth.type == "oneshot":
logger.info(f" oneshot auth found: {auth.oneshot}")
if current_time < auth.oneshot.ends_at:
if auth.oneshot.uses > 0:
decrementOneshot(db, auth.oneshot)
return True
logger.info("No more auths found")
return False
except exc.NoResultFound:
raise Exception("No Access with that key found, this might be a db error")
raise Exception("No AccessAuth with that key found!")
+111 -74
View File
@@ -1,33 +1,37 @@
import logging
logger = logging.getLogger(__name__)
import secrets
import threading
import time
import os
import secrets
from typing import Optional
from sqlmodel import Session
from dotenv import load_dotenv
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 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
import desfire.exceptions as desExceptions
from app.services.door import checkAccess, closeDoor, isDoorOpen, openDoor
from app.services.settings import settings
from app.services.door import openDoor, closeDoor, isDoorOpen, checkAccess
logger = logging.getLogger(__name__)
#ENV vars
load_dotenv()
MIFARE_APP_MASTER_KEY = os.getenv('MIFARE_APP_MASTER_KEY')
MIFARE_ACL_READ_BASE_KEY = os.getenv('MIFARE_ACL_READ_BASE_KEY')
MIFARE_ACL_WRITE_BASE_KEY = os.getenv('MIFARE_ACL_WRITE_BASE_KEY')
# 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
@@ -36,10 +40,15 @@ 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 == None:
logger.critical("NO MASTER KEY LOADED")
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="No key loaded! Check application.")
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()
@@ -48,51 +57,58 @@ def getCardService(timeout: int = 10):
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))
# 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
# 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)
assert applications[0] == get_list(MIFARE_APP_ID)
except AssertionError:
logger.error("No application found!")
logger.error("No application found! The card must be registered first!")
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
# 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
# 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)
@@ -107,24 +123,16 @@ def DeleteCard():
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:
logger.debug("Auth2")
desfire.authenticate(0x0, aes_null_key)
except:
logger.debug("Auth3")
desfire.authenticate(0x0, desKey)
desfire.select_application(0x0)
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")
raise HTTPException(
status_code=status.HTTP_410_GONE, detail="Card is empty."
)
desfire.select_application(MIFARE_APP_ID)
desfire.authenticate(0x0, aes_master_key)
@@ -136,14 +144,18 @@ def DeleteCard():
pass
scannerThread.start()
return rdata
except(Exception, AssertionError) as e:
except (Exception, AssertionError) as e:
logger.error(f"Error in deletion function: {e}", exc_info=True)
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error: {e}"
)
def WriteNewCard():
try:
checkForKey()
from app.main import scanner as scannerThread
scannerThread.stop()
cardservice = getCardService(20)
@@ -159,10 +171,17 @@ def WriteNewCard():
# Authenticate with default DES key
logger.debug("Authenticating with default DES key...")
desfire.authenticate(0x0, desKey)
#get uid
# get uid
uid = desfire.get_real_uid()
applications = desfire.get_application_ids()
logger.debug(f"Applications: {applications}")
if len(applications) >= 1:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="This Card already has a key!"
)
# Set default key
logger.debug("Setting default key...")
desfire.change_default_key(aes_null_key, 0x0)
@@ -189,7 +208,7 @@ def WriteNewCard():
# Select application
desfire.select_application(MIFARE_APP_ID)
#recreate key object
# recreate key object
desfire.authenticate(0x0, aes_null_key)
desfire.change_key(0x0, aes_null_key, aes_master_key, 0x1)
@@ -198,18 +217,28 @@ def WriteNewCard():
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)
# 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)
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)
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(
@@ -226,25 +255,29 @@ def WriteNewCard():
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))
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=":")
return key, to_hex_string(data=uid, separator=":")
except Exception as e:
logger.error(f"Error in write function: {e}", exc_info=True)
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error: {e}")
logger.error(f"Error in write function: {e}", exc_info=False)
scannerThread.start()
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error: {e}"
)
class BackgroundScanner:
def __init__(self, db):
self.db = db
self.is_running = False
self.thread: Optional[threading.Thread] = None
self.thread: threading.Thread | None = None
def start(self):
if self.is_running:
@@ -274,7 +307,7 @@ class BackgroundScanner:
logger.debug("READY after timout")
except Exception as e:
logger.error(f"Error in scan function: {e}", exc_info=True)
logger.error(f"Error in function _scan_loop: {e}", exc_info=False)
time.sleep(6)
def _read_card(self):
@@ -290,13 +323,17 @@ class BackgroundScanner:
rdata = readFileOnCard(desfire=desfire)
return rdata
except Exception as e:
logger.error(f"something went wrong: {e}")
logger.error(f"Error in function _read_card: {e}", exc_info=False)
time.sleep(5)
def _check_db(self, key):
check = checkAccess(key, self.db)
if check == True:
openDoor()
logger.info("Access granted!")
if isDoorOpen():
closeDoor()
logger.info("Door closed by key %s", key)
else:
logger.error("Access denied!")
check = checkAccess(key, self.db)
if check:
openDoor()
logger.info("Access granted!")
else:
logger.error("Access denied!")
+70
View File
@@ -0,0 +1,70 @@
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
@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
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()
+8 -2
View File
@@ -65,6 +65,10 @@
buildInputs = (old.buildInputs or []) ++ [ pkgs.pcsclite.dev ];
NIX_CFLAGS_COMPILE = "-I${pkgs.pcsclite.dev}/include/PCSC";
});
lgpio = prev.lgpio.overrideAttrs (old: {
nativeBuildInputs = (old.nativeBuildInputs or []) ++ [ pkgs.swig ];
buildInputs = (old.buildInputs or []) ++ [ pkgs.lgpio ];
});
})
]
)
@@ -93,7 +97,7 @@
UV_NO_SYNC = "1";
UV_PYTHON = pythonSet.python.interpreter;
UV_PYTHON_DOWNLOADS = "never";
LD_LIBRARY_PATH = "${lib.getLib pkgs.pcsclite}/lib";
LD_LIBRARY_PATH = "${lib.getLib pkgs.pcsclite}/lib:${lib.getLib pkgs.lgpio}/lib";
};
shellHook = ''
unset PYTHONPATH
@@ -119,9 +123,11 @@
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 "$@";
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; });
};
}
+74
View File
@@ -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;
};
};
};
}
+33 -2
View File
@@ -11,11 +11,19 @@ dependencies = [
"python-desfire",
"pyjwt[crypto]>=2.12.1",
"pwdlib[argon2]>=0.3.0",
"pytest>=9.0.3",
"requests>=2.33.1",
"pytest-cov>=7.1.0",
"setuptools>=82.0.1",
"pyscard>=2.3.1",
"alembic>=1.18.5",
"pydantic-settings>=2.13.1",
"paho-mqtt>=2.1.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]
@@ -24,3 +32,26 @@ 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"]
-164
View File
@@ -1,164 +0,0 @@
"""
This is a more involved example that performs initial configuration (often called personalization) of a DESFire card.
It performs the following steps:
1. Authenticate with the default DES key
3. Change the default key
2. Create an application
4. Change the application master key
6. Create a read and write key (diversified)
7. Create an encrypted file
8. Write the UID to the encrypted file
"""
import logging
import os
from smartcard.CardRequest import CardRequest
from smartcard.CardType import AnyCardType
from smartcard.Exceptions import CardRequestTimeoutException
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 dotenv import load_dotenv
# Please make sure to yet your own keys here before running this script
load_dotenv()
MIFARE_APP_MASTER_KEY = os.getenv('MIFARE_APP_MASTER_KEY')
MIFARE_ACL_READ_BASE_KEY = os.getenv('MIFARE_ACL_READ_BASE_KEY')
MIFARE_ACL_WRITE_BASE_KEY = os.getenv('MIFARE_ACL_WRITE_BASE_KEY')
# 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
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
cardtype = AnyCardType()
cardrequest = CardRequest(timeout=30, cardType=cardtype)
print("Please present DESfire tag...")
try:
cardservice = cardrequest.waitforcard()
except CardRequestTimeoutException:
print("No tag detected within the timeout.")
raise
cardservice.connection.connect()
# Create Desfire object
desfire = DESFire(PCSCDevice(cardservice.connection.component))
# Create Key objects
AES_NULL_KEY_DATA = "00" * 16
aes_keysettings = KeySettings(
key_type=DESFireKeyType.DF_KEY_AES,
)
aes_null_key = DESFireKey(aes_keysettings, AES_NULL_KEY_DATA)
# Authenticate with default DES key
print("Authenticating with default DES key...")
key_settings = desfire.get_key_setting()
mk = DESFireKey(key_settings, "00" * 8)
desfire.authenticate(0x0, mk)
# Get real UID
print("Getting real UID...")
uid = desfire.get_real_uid()
print(" - UID: ", to_hex_string(uid))
# Set default key
print("Setting default key...")
desfire.change_default_key(aes_null_key, 0x0)
# Create application
print("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)
print(" - Application created successfully.")
# Select application
print("Selecting application...")
desfire.select_application(MIFARE_APP_ID)
# Authenticate with AES key, as this has been set as the default key
print("Authenticating with AES key...")
# Create a new one as key data would be overriden by session data
aes_null_auth_key = DESFireKey(aes_keysettings, AES_NULL_KEY_DATA)
desfire.authenticate(0x0, aes_null_auth_key)
# Change Application master key
print("Changing application master key (AMK)...")
aes_app_mk = DESFireKey(aes_keysettings, MIFARE_APP_MASTER_KEY)
desfire.change_key(0x0, aes_null_key, aes_app_mk, 0x1)
# Re-Authenticate with new AES key
print("Re-authenticating with new AES key...")
desfire.authenticate(0x0, aes_app_mk)
# Change file read and write keys (diversified)
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)
print("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)
print("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)
print("Create encrypted file containing UID...")
file_settings = FileSettings(
file_size=8,
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)
print("Read and verify file settings again...")
file_data = desfire.get_file_settings(MIFARE_ENCRYPTED_FILE_ID)
assert file_data.file_size == 8
assert file_data.encryption == DESFireCommunicationMode.ENCRYPTED
assert file_data.permissions is not None
assert file_data.permissions.read_access == MIFARE_ACL_READ_BASE_KEY_ID
assert file_data.permissions.write_access == MIFARE_ACL_WRITE_BASE_KEY_ID
assert file_data.file_type == DESFireFileType.MDFT_STANDARD_DATA_FILE
print(" - File created successfully.")
print("Writing UID to encrypted file...")
data = [0x0] + uid
assert len(data) == 8
desfire.write_file_data(MIFARE_ENCRYPTED_FILE_ID, 0x0, file_data.encryption, get_list(data))
print("Reading from encrypted file...")
rdata = desfire.read_file_data(MIFARE_ENCRYPTED_FILE_ID, file_data)
assert rdata == data
print(" - Data written successfully.")
print("Personalization finished.")
+37 -19
View File
@@ -1,17 +1,32 @@
import os
from datetime import time
import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, create_engine, SQLModel
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from sqlmodel import Session, SQLModel, create_engine
os.environ["SECRET_KEY"] = "ff" * 16
from app.main import app
from app.model.models import UserDB, Card, GroupDB, AccessAuthorizationDB, Timetable, AaGroupLink
from app.model.models import (
AccessAuthorizationDB,
Card,
GroupDB,
Timetable,
UserDB,
)
from app.services.database import get_session
# Use in-memory SQLite for testing
TEST_SQLALCHEMY_DATABASE_URL = "sqlite://"
engine = create_engine(TEST_SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}, poolclass=StaticPool)
engine = create_engine(
TEST_SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
@pytest.fixture(scope="function")
def db_session():
@@ -25,6 +40,7 @@ def db_session():
@pytest.fixture(scope="function")
def client(db_session):
"""Create a test client with a database session override."""
def override_get_session():
yield db_session
@@ -38,10 +54,9 @@ def client(db_session):
def admin_user(db_session):
"""Create an admin user for testing."""
from app.services.auth import get_password_hash
admin = UserDB(
name="admin",
passwordhash=get_password_hash("admin123"),
is_admin=True
name="admin", passwordhash=get_password_hash("admin123"), is_admin=True
)
db_session.add(admin)
db_session.commit()
@@ -53,10 +68,9 @@ def admin_user(db_session):
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
name="user", passwordhash=get_password_hash("user123"), is_admin=False
)
db_session.add(user)
db_session.commit()
@@ -68,8 +82,7 @@ def regular_user(db_session):
def auth_headers(client, admin_user):
"""Get authentication headers for admin user."""
response = client.post(
"/api/v1/token",
data={"username": admin_user.name, "password": "admin123"}
"/api/v1/token", data={"username": admin_user.name, "password": "admin123"}
)
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
@@ -79,8 +92,7 @@ def auth_headers(client, admin_user):
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"}
"/api/v1/token", data={"username": regular_user.name, "password": "user123"}
)
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
@@ -99,7 +111,13 @@ def test_group(db_session):
@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")
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)
@@ -107,11 +125,11 @@ def test_card(db_session, test_group):
@pytest.fixture
def test_aa(db_session):
"""Create a test access authorization."""
def test_aa_tt(db_session):
"""Create a test access authorization with timetable."""
tt = Timetable(weekday=1, starttime=time(1, 0, 0, 0), duration=50)
aa = AccessAuthorizationDB(
name="Test AA",
is_active=True
name="Test AA", is_active=True, type="timetable", timetables=[tt]
)
db_session.add(aa)
db_session.commit()
+2
View File
@@ -4,9 +4,11 @@ def test_app_startup(client):
# Application should respond (even if it's a 404)
assert response.status_code in [404, 200]
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
+18 -11
View File
@@ -1,11 +1,18 @@
import pytest
import datetime
from app.model.models import (
UserBase, UserResponse, UserCreate, UserDB, UserUpdate,
GroupBase, GroupCreate, GroupDB, GroupResponse,
AccessAuthorizationBase, AccessAuthorizationCreate,
AccessAuthorizationDB, AccessAuthorizationResponse, AccessAuthorizationUpdate,
Card, Timetable, TimetableCreate, Token, TokenData, AaGroupLink
AaGroupLink,
AccessAuthorizationBase,
AccessAuthorizationCreate,
Card,
GroupBase,
GroupCreate,
TimetableCreate,
Token,
TokenData,
UserBase,
UserCreate,
UserUpdate,
)
@@ -18,7 +25,9 @@ def test_user_models():
assert user_base.is_admin is False
# Test UserCreate
user_create = UserCreate(name="New User", email="new@example.com", password="secret123")
user_create = UserCreate(
name="New User", email="new@example.com", password="secret123"
)
assert user_create.password == "secret123"
# Test UserUpdate
@@ -41,16 +50,14 @@ def test_group_models():
def test_access_authorization_models():
"""Test access authorization model creation and validation."""
# Test AccessAuthorizationBase
aa_base = AccessAuthorizationBase(name="Test AA", is_active=True)
aa_base = AccessAuthorizationBase(name="Test AA", is_active=True, type="timetable")
assert aa_base.name == "Test AA"
assert aa_base.is_active is True
# Test AccessAuthorizationCreate with timetables
timetable_create = TimetableCreate(weekday=1, starttime="08:00", duration=60)
aa_create = AccessAuthorizationCreate(
name="New AA",
is_active=False,
timetables=[timetable_create]
name="New AA", is_active=False, type="timetable", timetables=[timetable_create]
)
assert aa_create.name == "New AA"
assert aa_create.is_active is False
+84 -55
View File
@@ -1,29 +1,59 @@
import pytest
from fastapi import status
def test_create_access_auth(client, auth_headers):
def test_create_access_auth_tt(client, auth_headers):
"""Test creating a new access authorization."""
aa_data = {
"name": "New AA",
"name": "New tt_AA",
"type": "timetable",
"is_active": True,
"timetables": [
{"weekday": 1, "starttime": "08:00", "duration": 60},
{"weekday": 2, "starttime": "09:00", "duration": 90}
]
{"weekday": 2, "starttime": "09:00", "duration": 90},
],
}
response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert data["name"] == "New AA"
assert data["name"] == "New tt_AA"
assert data["is_active"] is True
assert data["type"] == "timetable"
assert "id" in data
assert len(data["timetables"]) == 2
def test_get_all_access_auths(client, auth_headers, test_aa):
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_get_all_access_auths(client, auth_headers, test_aa_tt):
"""Test retrieving all access authorizations."""
response = client.get("/api/v1/aa/", headers=auth_headers)
assert response.status_code == 200
@@ -32,17 +62,17 @@ def test_get_all_access_auths(client, auth_headers, test_aa):
assert len(aa_list) >= 1
aa_names = [aa["name"] for aa in aa_list]
assert test_aa.name in aa_names
assert test_aa_tt.name in aa_names
def test_get_access_auth_by_id(client, auth_headers, test_aa):
def test_get_access_auth_by_id(client, auth_headers, test_aa_tt):
"""Test retrieving a specific access authorization by ID."""
response = client.get(f"/api/v1/aa/{test_aa.id}", headers=auth_headers)
response = client.get(f"/api/v1/aa/{test_aa_tt.id}", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert data["id"] == test_aa.id
assert data["name"] == test_aa.name
assert data["id"] == test_aa_tt.id
assert data["name"] == test_aa_tt.name
def test_get_nonexistent_access_auth(client, auth_headers):
@@ -51,11 +81,10 @@ def test_get_nonexistent_access_auth(client, auth_headers):
assert response.status_code == 404
def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa):
def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa_tt):
"""Test assigning an access authorization to a group."""
response = client.put(
f"/api/v1/aa/assign/{test_group.id}/{test_aa.id}",
headers=auth_headers
f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers
)
assert response.status_code == 200
@@ -65,66 +94,68 @@ def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa):
# Note: The response model might not include the full relationship
def test_assign_already_assigned_access_auth(client, auth_headers, test_group, test_aa):
def test_assign_already_assigned_access_auth(
client, auth_headers, test_group, test_aa_tt
):
"""Test assigning an already assigned access authorization."""
# First assignment
client.put(f"/api/v1/aa/assign/{test_group.id}/{test_aa.id}", headers=auth_headers)
client.put(
f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.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.id}",
headers=auth_headers
f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.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):
def test_unassign_access_auth_from_group(client, auth_headers, test_group, test_aa_tt):
"""Test unassigning an access authorization from a group."""
# First assign
client.put(f"/api/v1/aa/assign/{test_group.id}/{test_aa.id}", headers=auth_headers)
client.put(
f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers
)
# Then unassign
response = client.put(
f"/api/v1/aa/unassign/{test_group.id}/{test_aa.id}",
headers=auth_headers
f"/api/v1/aa/unassign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers
)
assert response.status_code == 200
def test_unassign_nonexistent_assignment(client, auth_headers, test_group, test_aa):
def test_unassign_nonexistent_assignment(client, auth_headers, test_group, test_aa_tt):
"""Test unassigning a non-existent assignment."""
response = client.put(
f"/api/v1/aa/unassign/{test_group.id}/{test_aa.id}",
headers=auth_headers
f"/api/v1/aa/unassign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers
)
assert response.status_code == 404
def test_assign_to_nonexistent_group(client, auth_headers, test_aa):
def test_assign_to_nonexistent_group(client, auth_headers, test_aa_tt):
"""Test assigning an AA to a non-existent group."""
response = client.put(f"/api/v1/aa/assign/99999/{test_aa.id}", headers=auth_headers)
response = client.put(
f"/api/v1/aa/assign/99999/{test_aa_tt.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)
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):
def test_update_access_auth(client, auth_headers, test_aa_tt):
"""Test updating an access authorization."""
update_data = {
"name": "Updated AA",
"is_active": False
}
update_data = {"name": "Updated AA", "is_active": False}
response = client.patch(
f"/api/v1/aa/{test_aa.id}",
json=update_data,
headers=auth_headers
f"/api/v1/aa/{test_aa_tt.id}", json=update_data, headers=auth_headers
)
assert response.status_code == 200
@@ -133,18 +164,14 @@ def test_update_access_auth(client, auth_headers, test_aa):
assert data["is_active"] is False
def test_update_access_auth_with_timetables(client, auth_headers, test_aa):
def test_update_access_auth_with_timetables(client, auth_headers, test_aa_tt):
"""Test updating an access authorization with new timetables."""
update_data = {
"timetables": [
{"weekday": 5, "starttime": "10:00", "duration": 120}
]
"timetables": [{"weekday": 5, "starttime": "10:00", "duration": 120}]
}
response = client.patch(
f"/api/v1/aa/{test_aa.id}",
json=update_data,
headers=auth_headers
f"/api/v1/aa/{test_aa_tt.id}", json=update_data, headers=auth_headers
)
assert response.status_code == 200
jresponse = response.json()
@@ -161,14 +188,14 @@ def test_update_nonexistent_access_auth(client, auth_headers):
assert response.status_code == 404
def test_delete_access_auth(client, auth_headers, test_aa):
def test_delete_access_auth(client, auth_headers, test_aa_tt):
"""Test deleting an access authorization."""
response = client.delete(f"/api/v1/aa/{test_aa.id}", headers=auth_headers)
response = client.delete(f"/api/v1/aa/{test_aa_tt.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.id}", headers=auth_headers)
response = client.get(f"/api/v1/aa/{test_aa_tt.id}", headers=auth_headers)
assert response.status_code == 404
@@ -178,13 +205,13 @@ def test_delete_nonexistent_access_auth(client, auth_headers):
assert response.status_code == 404
def test_aa_operations_by_non_admin(client, test_aa, user_auth_headers):
def test_aa_tt_operations_by_non_admin(client, test_aa_tt, 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, "timetables": []},
headers=user_auth_headers
"/api/v1/aa/",
json={"name": "test", "is_active": True, "timetables": []},
headers=user_auth_headers,
)
assert response.status_code == 403
@@ -193,5 +220,7 @@ def test_aa_operations_by_non_admin(client, test_aa, user_auth_headers):
assert response.status_code == 403
# Try to assign AA
response = client.put(f"/api/v1/aa/assign/1/{test_aa.id}", headers=user_auth_headers)
response = client.put(
f"/api/v1/aa/assign/1/{test_aa_tt.id}", headers=user_auth_headers
)
assert response.status_code == 403
+22 -15
View File
@@ -1,12 +1,19 @@
from datetime import timedelta
import pytest
from datetime import datetime, timedelta, timezone
from fastapi import HTTPException, status
from app.services.auth import (
verify_password, get_password_hash, get_user, authenticate_user,
create_access_token, get_current_user, auth_is_admin, create_first_user
)
from app.model.models import UserDB
from jwt.exceptions import InvalidTokenError
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():
@@ -84,7 +91,7 @@ def test_create_access_token():
def test_get_current_user(db_session, admin_user):
"""Test getting current user from token."""
from app.services.auth import create_access_token, get_current_user
from app.services.auth import create_access_token
# Create token for admin user
token = create_access_token(data={"sub": admin_user.name})
@@ -102,7 +109,9 @@ def test_get_current_user(db_session, admin_user):
# 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)
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)
@@ -111,7 +120,7 @@ def test_get_current_user(db_session, admin_user):
def test_auth_is_admin(db_session, admin_user, regular_user):
"""Test admin authorization check."""
from app.services.auth import create_access_token, auth_is_admin
from app.services.auth import create_access_token
# Create token for admin user
admin_token = create_access_token(data={"sub": admin_user.name})
@@ -133,6 +142,7 @@ 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)
@@ -158,8 +168,7 @@ 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"}
"/api/v1/token", data={"username": admin_user.name, "password": "admin123"}
)
assert response.status_code == 200
data = response.json()
@@ -168,15 +177,13 @@ def test_token_endpoint(client, admin_user):
# Test failed login with wrong password
response = client.post(
"/api/v1/token",
data={"username": admin_user.name, "password": "wrongpassword"}
"/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"}
"/api/v1/token", data={"username": "nonexistent", "password": "password"}
)
assert response.status_code == 401
+11 -24
View File
@@ -1,6 +1,3 @@
import pytest
from fastapi import status
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)
@@ -23,53 +20,43 @@ def test_get_cards_for_nonexistent_group(client, auth_headers):
def test_card_operations_by_non_admin(client, test_group, user_auth_headers):
"""Test that non-admin users cannot perform card operations."""
# Try to add a card
response = client.post(f"/api/v1/cards/", headers=user_auth_headers)
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"
}
update_data = {"name": "changed_name", "enabled": "False"}
response = client.patch(
f"/api/v1/cards/{test_card.id}",
json=update_data,
headers=auth_headers
f"/api/v1/cards/{test_card.id}", json=update_data, headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert data["name"] == "changed_name"
assert data["enabled"] == False
assert data["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(
f"/api/v1/cards/9999",
json={},
headers=auth_headers
)
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"
}
update_data = {"group_id": "9999"}
response = client.patch(
f"/api/v1/cards/{test_card.id}",
json=update_data,
headers=auth_headers
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"]
assert "GroupID not found" in response.json()["detail"]
+7 -6
View File
@@ -1,17 +1,18 @@
import pytest
from sqlmodel import Session, select
from app.services.database import create_db_and_tables, get_session, add_and_refresh
from app.model.models import UserDB, GroupDB, Card
from sqlmodel import Session
from app.model.models import UserDB
from app.services.database import add_and_refresh, create_db_and_tables
def test_create_db_and_tables():
"""Test database and tables creation."""
# This is primarily an integration test
from sqlalchemy import inspect
from app.services.database import engine
from app.services.database import get_engine
create_db_and_tables()
inspector = inspect(engine)
inspector = inspect(get_engine())
# Check that tables exist
tables = inspector.get_table_names()
+170 -12
View File
@@ -1,7 +1,16 @@
import pytest
import datetime
import pytest
from app.model.models import (
AccessAuthorizationDB,
Card,
GroupDB,
OneShotAccess,
Timetable,
)
from app.services.door import checkAccess
from app.model.models import Card, GroupDB, AccessAuthorizationDB, Timetable
def test_check_access_with_valid_timetable(db_session):
# Setup: create card with valid access
@@ -9,17 +18,23 @@ def test_check_access_with_valid_timetable(db_session):
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")
card = Card(
key="test-key-123",
group_id=group.id,
enabled=True,
name="test_card",
card_serial="00:00:00:00:00:00:00",
)
db_session.add(card)
timetable = Timetable(
weekday=datetime.datetime.weekday(datetime.date.today()),
starttime=datetime.datetime.now().time(),
duration=120 # 2 hours
duration=120, # 2 hours
)
db_session.add(timetable)
aa = AccessAuthorizationDB(name="Test AA", is_active=True)
aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="timetable")
db_session.add(aa)
aa.timetables = [timetable]
group.accessauths = [aa]
@@ -27,8 +42,8 @@ def test_check_access_with_valid_timetable(db_session):
db_session.commit()
# Test: access should be granted within time window
result = checkAccess("test-key-123", db_session)
assert result == True
assert checkAccess("test-key-123", db_session)
def test_check_access_outside_hours(db_session):
# Test when current time is outside valid hours
@@ -36,26 +51,169 @@ def test_check_access_outside_hours(db_session):
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")
card = Card(
key="test-key-123",
group_id=group.id,
enabled=True,
name="test_card",
card_serial="00:00:00:00:00:00:00",
)
db_session.add(card)
timetable = Timetable(
weekday=datetime.datetime.weekday(datetime.date.today()),
starttime=datetime.time(1, 0),
duration=1 # 2 hours
duration=1, # 2 hours
)
db_session.add(timetable)
aa = AccessAuthorizationDB(name="Test AA", is_active=True)
aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="timetable")
db_session.add(aa)
aa.timetables = [timetable]
group.accessauths = [aa]
db_session.commit()
result = checkAccess("test-key-123", db_session)
assert result == False
assert not checkAccess("test-key-123", db_session)
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):
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)
timetable = Timetable(
weekday=datetime.datetime.weekday(datetime.date.today()),
starttime=datetime.datetime.now().time(),
duration=120, # 2 hours
)
db_session.add(timetable)
aa = AccessAuthorizationDB(name="Test AA", is_active=False, type="timetable")
db_session.add(aa)
aa.timetables = [timetable]
group.accessauths = [aa]
db_session.commit()
assert checkAccess("test-key-123", db_session) == False
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)
timetable = Timetable(
weekday=datetime.datetime.weekday(datetime.date.today()),
starttime=datetime.datetime.now().time(),
duration=120, # 2 hours
)
db_session.add(timetable)
aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="timetable")
db_session.add(aa)
aa.timetables = [timetable]
group.accessauths = [aa]
db_session.commit()
assert checkAccess("test-key-123", db_session) == False
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 timetable with inactive AA
timetable1 = Timetable(
weekday=datetime.datetime.weekday(datetime.date.today()),
starttime=datetime.datetime.now().time(),
duration=120, # 2 hours
)
db_session.add(timetable1)
aa1 = AccessAuthorizationDB(name="First AA", is_active=False, type="timetable")
db_session.add(aa1)
aa1.timetables = [timetable1]
#Creating timetable with active AA
timetable2 = Timetable(
weekday=datetime.datetime.weekday(datetime.date.today()),
starttime=datetime.datetime.now().time(),
duration=120, # 2 hours
)
db_session.add(timetable2)
aa2 = AccessAuthorizationDB(name="Second AA", is_active=True, type="timetable")
db_session.add(aa2)
aa2.timetables = [timetable2]
group.accessauths = [aa1, aa2]
db_session.commit()
assert checkAccess("test-key-123", db_session) == True
+1 -7
View File
@@ -1,7 +1,3 @@
import pytest
from fastapi import status
def test_create_group(client, auth_headers):
"""Test creating a new group."""
group_data = {"name": "New Test Group"}
@@ -57,9 +53,7 @@ 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
"/api/v1/groups/", json={"name": "test"}, headers=user_auth_headers
)
assert response.status_code == 403
+15 -24
View File
@@ -1,14 +1,10 @@
import pytest
from fastapi import status
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"
"password": "newpassword123",
}
response = client.post("/api/v1/users/", json=user_data, headers=auth_headers)
@@ -27,7 +23,7 @@ def test_create_user_unauthorized(client):
user_data = {
"name": "unauthorized_user",
"email": "unauthorized@example.com",
"password": "password123"
"password": "password123",
}
response = client.post("/api/v1/users/", json=user_data)
@@ -56,6 +52,7 @@ def test_get_user_by_id(client, auth_headers, regular_user):
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)
@@ -66,6 +63,7 @@ def test_get_current_user(client, auth_headers, admin_user):
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)
@@ -75,15 +73,10 @@ def test_get_nonexistent_user(client, auth_headers):
def test_update_user(client, auth_headers, regular_user):
"""Test updating a user."""
update_data = {
"name": "updated_name",
"email": "updated@example.com"
}
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
f"/api/v1/users/{regular_user.id}", json=update_data, headers=auth_headers
)
assert response.status_code == 200
@@ -96,21 +89,17 @@ def test_update_user(client, auth_headers, regular_user):
def test_update_user_password(client, auth_headers, regular_user):
"""Test updating a user's password."""
update_data = {
"password": "new_password_456"
}
update_data = {"password": "new_password_456"}
response = client.patch(
f"/api/v1/users/{regular_user.id}",
json=update_data,
headers=auth_headers
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"}
data={"username": regular_user.name, "password": "new_password_456"},
)
assert login_response.status_code == 200
@@ -118,7 +107,9 @@ def test_update_user_password(client, auth_headers, regular_user):
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)
response = client.patch(
"/api/v1/users/99999", json=update_data, headers=auth_headers
)
assert response.status_code == 404
@@ -143,9 +134,9 @@ 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
"/api/v1/users/",
json={"name": "test", "password": "pass"},
headers=user_auth_headers,
)
assert response.status_code == 403
Generated
+210 -4
View File
@@ -6,6 +6,20 @@ resolution-markers = [
"python_full_version < '3.14'",
]
[[package]]
name = "alembic"
version = "1.18.5"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "mako" },
{ name = "sqlalchemy" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/1a/cc/ac0bed8e562e7407fe55c3ba85a4dce86e6dbd8730887bd1e406a6c5c18a/alembic-1.18.5.tar.gz", hash = "sha256:1554982221dd17e9a749b53902407578eb305e453f71999e8c7f0a48389fff8e", size = 2060480, upload-time = "2026-06-25T15:20:54.888Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/96/78/5fe6dc3a3a5b2f5a2a4faef8bfe336d5fa049a38884ab3172e0098160c01/alembic-1.18.5-py3-none-any.whl", hash = "sha256:06d8ba9d04558022f5395e9317de03d270f3dced49cee01f89fe7a13c26f14bc", size = 264664, upload-time = "2026-06-25T15:20:56.673Z" },
]
[[package]]
name = "annotated-doc"
version = "0.0.4"
@@ -79,6 +93,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/42/b9/f8d6fa329ab25128b7e98fd83a3cb34d9db5b059a9847eddb840a0af45dd/argon2_cffi_bindings-25.1.0-cp39-abi3-win_arm64.whl", hash = "sha256:b0fdbcf513833809c882823f98dc2f931cf659d9a1429616ac3adebb49f5db94", size = 27149, upload-time = "2025-07-30T10:01:59.329Z" },
]
[[package]]
name = "attrs"
version = "26.1.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/9a/8e/82a0fe20a541c03148528be8cac2408564a6c9a0cc7e9171802bc1d26985/attrs-26.1.0.tar.gz", hash = "sha256:d03ceb89cb322a8fd706d4fb91940737b6642aa36998fe130a9bc96c985eff32", size = 952055, upload-time = "2026-03-19T14:22:25.026Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/64/b4/17d4b0b2a2dc85a6df63d1157e028ed19f90d4cd97c36717afef2bc2f395/attrs-26.1.0-py3-none-any.whl", hash = "sha256:c647aa4a12dfbad9333ca4e71fe62ddc36f4e63b2d260a37a8b83d2f043ac309", size = 67548, upload-time = "2026-03-19T14:22:23.645Z" },
]
[[package]]
name = "build"
version = "1.4.3"
@@ -111,6 +134,19 @@ filecache = [
{ name = "filelock" },
]
[[package]]
name = "cattrs"
version = "26.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "attrs" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/a0/ec/ba18945e7d6e55a58364d9fb2e46049c1c2998b3d805f19b703f14e81057/cattrs-26.1.0.tar.gz", hash = "sha256:fa239e0f0ec0715ba34852ce813986dfed1e12117e209b816ab87401271cdd40", size = 495672, upload-time = "2026-02-18T22:15:19.406Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/80/56/60547f7801b97c67e97491dc3d9ade9fbccbd0325058fd3dfcb2f5d98d90/cattrs-26.1.0-py3-none-any.whl", hash = "sha256:d1e0804c42639494d469d08d4f26d6b9de9b8ab26b446db7b5f8c2e97f7c3096", size = 73054, upload-time = "2026-02-18T22:15:17.958Z" },
]
[[package]]
name = "certifi"
version = "2026.2.25"
@@ -405,6 +441,19 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/ba/5a/18ad964b0086c6e62e2e7500f7edc89e3faa45033c71c1893d34eed2b2de/dnspython-2.8.0-py3-none-any.whl", hash = "sha256:01d9bbc4a2d76bf0db7c1f729812ded6d912bd318d3b1cf81d30c0f845dbf3af", size = 331094, upload-time = "2025-09-07T18:57:58.071Z" },
]
[[package]]
name = "docstring-to-markdown"
version = "0.17"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "importlib-metadata" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/52/d8/8abe80d62c5dce1075578031bcfde07e735bcf0afe2886dd48b470162ab4/docstring_to_markdown-0.17.tar.gz", hash = "sha256:df72a112294c7492487c9da2451cae0faeee06e86008245c188c5761c9590ca3", size = 32260, upload-time = "2025-05-02T15:09:07.932Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/56/7b/af3d0da15bed3a8665419bb3a630585756920f4ad67abfdfef26240ebcc0/docstring_to_markdown-0.17-py3-none-any.whl", hash = "sha256:fd7d5094aa83943bf5f9e1a13701866b7c452eac19765380dead666e36d3711c", size = 23479, upload-time = "2025-05-02T15:09:06.676Z" },
]
[[package]]
name = "dulwich"
version = "1.1.0"
@@ -608,34 +657,54 @@ name = "gatekeeper"
version = "0.1.0"
source = { virtual = "." }
dependencies = [
{ name = "alembic" },
{ name = "fastapi", extra = ["standard"] },
{ name = "lgpio", marker = "platform_machine == 'aarch64' and sys_platform == 'linux'" },
{ name = "paho-mqtt" },
{ name = "poetry" },
{ name = "pwdlib", extra = ["argon2"] },
{ name = "pydantic-settings" },
{ name = "pyjwt", extra = ["crypto"] },
{ name = "pyscard" },
{ name = "pytest" },
{ name = "pytest-cov" },
{ name = "python-desfire" },
{ name = "requests" },
{ name = "setuptools" },
{ name = "sqlmodel" },
]
[package.dev-dependencies]
dev = [
{ name = "jedi-language-server" },
{ name = "pytest" },
{ name = "pytest-cov" },
{ name = "ruff" },
]
[package.metadata]
requires-dist = [
{ name = "alembic", specifier = ">=1.18.5" },
{ name = "fastapi", extras = ["standard"], specifier = ">=0.135.3" },
{ name = "lgpio", marker = "platform_machine == 'aarch64' and sys_platform == 'linux'", specifier = ">=0.2.2.0" },
{ name = "paho-mqtt", specifier = ">=2.1.0" },
{ name = "poetry", specifier = ">=2.3.4" },
{ name = "pwdlib", extras = ["argon2"], specifier = ">=0.3.0" },
{ name = "pydantic-settings", specifier = ">=2.13.1" },
{ name = "pyjwt", extras = ["crypto"], specifier = ">=2.12.1" },
{ name = "pyscard", specifier = ">=2.3.1" },
{ name = "pytest", specifier = ">=9.0.3" },
{ name = "pytest-cov", specifier = ">=7.1.0" },
{ name = "python-desfire", git = "https://github.com/waza-ari/python-desfire" },
{ name = "requests", specifier = ">=2.33.1" },
{ name = "setuptools", specifier = ">=82.0.1" },
{ name = "sqlmodel", specifier = ">=0.0.38" },
]
[package.metadata.requires-dev]
dev = [
{ name = "jedi-language-server", specifier = ">=0.47.0" },
{ name = "pytest", specifier = ">=9.0.3" },
{ name = "pytest-cov", specifier = ">=7.1.0" },
{ name = "ruff", specifier = ">=0.16.0" },
]
[[package]]
name = "greenlet"
version = "3.4.0"
@@ -735,6 +804,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/0e/61/66938bbb5fc52dbdf84594873d5b51fb1f7c7794e9c0f5bd885f30bc507b/idna-3.11-py3-none-any.whl", hash = "sha256:771a87f49d9defaf64091e6e6fe9c18d4833f140bd19464795bc32d966ca37ea", size = 71008, upload-time = "2025-10-12T14:55:18.883Z" },
]
[[package]]
name = "importlib-metadata"
version = "9.0.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "zipp" },
]
sdist = { url = "https://files.pythonhosted.org/packages/a9/01/15bb152d77b21318514a96f43af312635eb2500c96b55398d020c93d86ea/importlib_metadata-9.0.0.tar.gz", hash = "sha256:a4f57ab599e6a2e3016d7595cfd72eb4661a5106e787a95bcc90c7105b831efc", size = 56405, upload-time = "2026-03-20T06:42:56.999Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/38/3d/2d244233ac4f76e38533cfcb2991c9eb4c7bf688ae0a036d30725b8faafe/importlib_metadata-9.0.0-py3-none-any.whl", hash = "sha256:2d21d1cc5a017bd0559e36150c21c830ab1dc304dedd1b7ea85d20f45ef3edd7", size = 27789, upload-time = "2026-03-20T06:42:55.665Z" },
]
[[package]]
name = "iniconfig"
version = "2.3.0"
@@ -786,6 +867,34 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/fd/c4/813bb09f0985cb21e959f21f2464169eca882656849adf727ac7bb7e1767/jaraco_functools-4.4.0-py3-none-any.whl", hash = "sha256:9eec1e36f45c818d9bf307c8948eb03b2b56cd44087b3cdc989abca1f20b9176", size = 10481, upload-time = "2025-12-21T09:29:42.27Z" },
]
[[package]]
name = "jedi"
version = "0.20.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "parso" },
]
sdist = { url = "https://files.pythonhosted.org/packages/46/b7/a3635f6a2d7cf5b5dd98064fc1d5fbbafcb25477bcea204a3a92145d158b/jedi-0.20.0.tar.gz", hash = "sha256:c3f4ccbd276696f4b19c54618d4fb18f9fc24b0aef02acf704b23f487daa1011", size = 3119416, upload-time = "2026-05-01T23:38:47.814Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/9a/93/242e2eab5fe682ffcb8b0084bde703a41d51e17ee0f3a31ff0d9d813620a/jedi-0.20.0-py2.py3-none-any.whl", hash = "sha256:7bdd9c2634f56713299976f4cbd59cb3fa92165cc5e05ea811fb253480728b67", size = 4884812, upload-time = "2026-05-01T23:38:43.919Z" },
]
[[package]]
name = "jedi-language-server"
version = "0.47.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "cattrs" },
{ name = "docstring-to-markdown" },
{ name = "jedi" },
{ name = "lsprotocol" },
{ name = "pygls" },
]
sdist = { url = "https://files.pythonhosted.org/packages/f5/00/0cc9fc9996f79f1935a39236f300e6b2399aed34b8170248c5108e38e652/jedi_language_server-0.47.0.tar.gz", hash = "sha256:43d9b89d07b504470544b80d76739449856a2be6565571cb362ee0681f6f0b86", size = 91714, upload-time = "2026-05-31T14:02:43.312Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/05/e1/ad4afa632b4ea58f422391d11e86b5100b5578879af6bf04bf5aaf4beede/jedi_language_server-0.47.0-py3-none-any.whl", hash = "sha256:0e00d5af41c1a966c9ec73221e81b1af3b8c0243d29477e5441c661d5b61ec3e", size = 33844, upload-time = "2026-05-31T14:02:42.161Z" },
]
[[package]]
name = "jeepney"
version = "0.9.0"
@@ -824,6 +933,37 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/81/db/e655086b7f3a705df045bf0933bdd9c2f79bb3c97bfef1384598bb79a217/keyring-25.7.0-py3-none-any.whl", hash = "sha256:be4a0b195f149690c166e850609a477c532ddbfbaed96a404d4e43f8d5e2689f", size = 39160, upload-time = "2025-11-16T16:26:08.402Z" },
]
[[package]]
name = "lgpio"
version = "0.2.2.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/56/33/26ec2e8049eaa2f077bf23a12dc61ca559fbfa7bea0516bf263d657ae275/lgpio-0.2.2.0.tar.gz", hash = "sha256:11372e653b200f76a0b3ef8a23a0735c85ec678a9f8550b9893151ed0f863fff", size = 90087, upload-time = "2024-03-29T21:59:55.901Z" }
[[package]]
name = "lsprotocol"
version = "2025.0.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "attrs" },
{ name = "cattrs" },
]
sdist = { url = "https://files.pythonhosted.org/packages/e9/26/67b84e6ec1402f0e6764ef3d2a0aaf9a79522cc1d37738f4e5bb0b21521a/lsprotocol-2025.0.0.tar.gz", hash = "sha256:e879da2b9301e82cfc3e60d805630487ac2f7ab17492f4f5ba5aaba94fe56c29", size = 74896, upload-time = "2025-06-17T21:30:18.156Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/7b/f0/92f2d609d6642b5f30cb50a885d2bf1483301c69d5786286500d15651ef2/lsprotocol-2025.0.0-py3-none-any.whl", hash = "sha256:f9d78f25221f2a60eaa4a96d3b4ffae011b107537facee61d3da3313880995c7", size = 76250, upload-time = "2025-06-17T21:30:19.455Z" },
]
[[package]]
name = "mako"
version = "1.3.12"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "markupsafe" },
]
sdist = { url = "https://files.pythonhosted.org/packages/00/62/791b31e69ae182791ec67f04850f2f062716bbd205483d63a215f3e062d3/mako-1.3.12.tar.gz", hash = "sha256:9f778e93289bd410bb35daadeb4fc66d95a746f0b75777b942088b7fd7af550a", size = 400219, upload-time = "2026-04-28T19:01:08.512Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/bc/b1/a0ec7a5a9db730a08daef1fdfb8090435b82465abbf758a596f0ea88727e/mako-1.3.12-py3-none-any.whl", hash = "sha256:8f61569480282dbf557145ce441e4ba888be453c30989f879f0d652e39f53ea9", size = 78521, upload-time = "2026-04-28T19:01:10.393Z" },
]
[[package]]
name = "markdown-it-py"
version = "4.0.0"
@@ -950,6 +1090,24 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/7a/c2/920ef838e2f0028c8262f16101ec09ebd5969864e5a64c4c05fad0617c56/packaging-26.1-py3-none-any.whl", hash = "sha256:5d9c0669c6285e491e0ced2eee587eaf67b670d94a19e94e3984a481aba6802f", size = 95831, upload-time = "2026-04-14T21:12:47.56Z" },
]
[[package]]
name = "paho-mqtt"
version = "2.1.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/39/15/0a6214e76d4d32e7f663b109cf71fb22561c2be0f701d67f93950cd40542/paho_mqtt-2.1.0.tar.gz", hash = "sha256:12d6e7511d4137555a3f6ea167ae846af2c7357b10bc6fa4f7c3968fc1723834", size = 148848, upload-time = "2024-04-29T19:52:55.591Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c4/cb/00451c3cf31790287768bb12c6bec834f5d292eaf3022afc88e14b8afc94/paho_mqtt-2.1.0-py3-none-any.whl", hash = "sha256:6db9ba9b34ed5bc6b6e3812718c7e06e2fd7444540df2455d2c51bd58808feee", size = 67219, upload-time = "2024-04-29T19:52:48.345Z" },
]
[[package]]
name = "parso"
version = "0.8.7"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/30/4b/90c937815137d43ce71ba043cd3566221e9df6b9c805f24b5d138c9d40a7/parso-0.8.7.tar.gz", hash = "sha256:eaaac4c9fdd5e9e8852dc778d2d7405897ec510f2a298071453e5e3a07914bb1", size = 401824, upload-time = "2026-05-01T23:13:02.138Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/99/5d/8268b644392ee874ee82a635cd0df1773de230bde356c38de28e298392cc/parso-0.8.7-py2.py3-none-any.whl", hash = "sha256:a8926eb2a1b915486941fdbd31e86a4baf88fe8c210f25f2f35ecec5b574ca1c", size = 107025, upload-time = "2026-05-01T23:12:58.867Z" },
]
[[package]]
name = "pbs-installer"
version = "2026.4.7"
@@ -1191,6 +1349,20 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/00/4b/ccc026168948fec4f7555b9164c724cf4125eac006e176541483d2c959be/pydantic_settings-2.13.1-py3-none-any.whl", hash = "sha256:d56fd801823dbeae7f0975e1f8c8e25c258eb75d278ea7abb5d9cebb01b56237", size = 58929, upload-time = "2026-02-19T13:45:06.034Z" },
]
[[package]]
name = "pygls"
version = "2.1.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "attrs" },
{ name = "cattrs" },
{ name = "lsprotocol" },
]
sdist = { url = "https://files.pythonhosted.org/packages/da/2e/7bbe061d175c0baddde8fc9edb908a4c31ba5d9165b8c68e3439c3a9f138/pygls-2.1.1.tar.gz", hash = "sha256:1da03ba9053201bb337dcdd8d121df70feb2a91e1a0dcc74de5da79755b1a201", size = 55091, upload-time = "2026-03-25T11:19:10.541Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/fd/1a/208293b6c350f5abea6941d5606080d4a492644052504f5312e5de30a902/pygls-2.1.1-py3-none-any.whl", hash = "sha256:510a6dea2476177230c7d851125e5948efdf3fdb9ebfd8543fc434972f8faed4", size = 68975, upload-time = "2026-03-25T11:19:11.374Z" },
]
[[package]]
name = "pygments"
version = "2.20.0"
@@ -1509,6 +1681,31 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/79/62/b88e5879512c55b8ee979c666ee6902adc4ed05007226de266410ae27965/rignore-0.7.6-cp314-cp314t-win_arm64.whl", hash = "sha256:b83adabeb3e8cf662cabe1931b83e165b88c526fa6af6b3aa90429686e474896", size = 656035, upload-time = "2025-11-05T21:41:31.13Z" },
]
[[package]]
name = "ruff"
version = "0.16.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/4d/94/1e5e4967626faf12fa56999cd6222dff6992ceb086ad7945756baf70c7a7/ruff-0.16.0.tar.gz", hash = "sha256:e460aafd5495ec89efaa6ced2e4a9a581116451e1c88b9d37ef497e0f8e93982", size = 4790557, upload-time = "2026-07-23T19:11:30.981Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/4b/81/1c8818fee7ce1a04cd7d1b3172e0a8f8e4f1dc4feb7fc390e16daa8af323/ruff-0.16.0-py3-none-linux_armv6l.whl", hash = "sha256:e5115729eb08c585e5121978ba5d5b60caeae394ce21b9fb5e6cd33a1c6c9b1e", size = 10754633, upload-time = "2026-07-23T19:10:46.415Z" },
{ url = "https://files.pythonhosted.org/packages/23/df/beaf59c09d68db84304d555f188b276a77132a5d5b0b67a5c762aa143628/ruff-0.16.0-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:3c954b1d580bfa035b41654f7858cc7e71d5fc3ac5b723dd62bd9133830ed522", size = 10969164, upload-time = "2026-07-23T19:10:50.271Z" },
{ url = "https://files.pythonhosted.org/packages/42/ce/741cd197496a1abbf51352710fd15ed995d2a2be87189c1da26a450d6e83/ruff-0.16.0-py3-none-macosx_11_0_arm64.whl", hash = "sha256:e01c21d10eb1b29f47b7454e1f4056db9a3f0260c646aa88457c610291db9f81", size = 10488846, upload-time = "2026-07-23T19:10:52.639Z" },
{ url = "https://files.pythonhosted.org/packages/52/2a/a2db8e88cade358f5cdcb05674a917751074109315d014eb6352d9a893f7/ruff-0.16.0-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:6e364e5ed22ed8dc05082fd78e35308618260907ac2d3c1d637b2e682415b6c9", size = 10889729, upload-time = "2026-07-23T19:10:54.89Z" },
{ url = "https://files.pythonhosted.org/packages/42/65/62a771694ebd63029dc953e27dbad40e1588bd4860ff9fe881018fddaa49/ruff-0.16.0-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:d327b8fc113a1d4421a04f3839d3752057c8dd1ee320223a6f3f52d04ada462a", size = 10568275, upload-time = "2026-07-23T19:10:56.993Z" },
{ url = "https://files.pythonhosted.org/packages/3f/e2/ced249fe8af5f086c5c58cc21cc3356d50f32f7401c5df87050c999620a7/ruff-0.16.0-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:a9b50c55e263103586b3dcf5f73d479eb8cb5fdb6098fec59a62891dab653717", size = 11385112, upload-time = "2026-07-23T19:10:59.615Z" },
{ url = "https://files.pythonhosted.org/packages/87/0b/05154977a8fd69eeb6c103271f55403bfd8711f5c0f8ed07489d95a504e7/ruff-0.16.0-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:0ff4a79ce3ec0172f3241943835de1c4cb4e2dcd07f0f8c2d02603dbbbee4b17", size = 12207008, upload-time = "2026-07-23T19:11:02.154Z" },
{ url = "https://files.pythonhosted.org/packages/fb/29/98225831a3a1eab0e02f4acc6ca6559a98611dcc68b6965ff4b7234627c1/ruff-0.16.0-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:e95c448fca1fb2a18372a9440926c5a6ee789639bb975c72e7ae6d0b04218ab4", size = 11650842, upload-time = "2026-07-23T19:11:04.557Z" },
{ url = "https://files.pythonhosted.org/packages/91/66/6bd3cf90500653d55dc0ffc8507aa8300bd49d0214b2e8cb4d3fef2943ba/ruff-0.16.0-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:4f11a8d11010301d0a398a2fdef67691feca7294da6aef55e2150e8fa2cd520b", size = 11400718, upload-time = "2026-07-23T19:11:09.233Z" },
{ url = "https://files.pythonhosted.org/packages/8e/a2/a54eb4eae05d66364050a5d3b8a9c5ef88196531b3cbe7109d873f87f819/ruff-0.16.0-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:48044c678e9cb8698246c99b14aaccfa6601dea7379eb48a6f8f73f7a6d86cd0", size = 11426177, upload-time = "2026-07-23T19:11:11.994Z" },
{ url = "https://files.pythonhosted.org/packages/1a/be/16e3eea4b2a478a496919f5e36f17c4559e54620bd3bbac5d6affa068006/ruff-0.16.0-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:7aa0959bad8eb8bef50340154fc9b58678dae31fa4293afa38b44b6e552c0213", size = 10856126, upload-time = "2026-07-23T19:11:14.221Z" },
{ url = "https://files.pythonhosted.org/packages/a2/84/252eb8b868a16eec7257c14f504f77537e734b2d69c762e639e588e304a3/ruff-0.16.0-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:28ea2b7df8ebf7f9da6b7d47b230ab48f387c0a29be3b474c4d0740e197bb9af", size = 10571208, upload-time = "2026-07-23T19:11:16.378Z" },
{ url = "https://files.pythonhosted.org/packages/21/09/817a482f542f7570cbb4554b26e896610c7114f539b1d9e2d2145bf6bef6/ruff-0.16.0-py3-none-musllinux_1_2_i686.whl", hash = "sha256:33a3dfac8c35f81498dea9181bccc2f4c4bc8f1521a1dd9406e77643e0f0fb09", size = 11063329, upload-time = "2026-07-23T19:11:19.173Z" },
{ url = "https://files.pythonhosted.org/packages/2e/23/9403c180ca1cb9b1f7335f5c3e5305c09d49ea5b345196682a36028bde4a/ruff-0.16.0-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:a5237a0bda500d30d81b8e07a6973a5cbc772864cbf746ae2f4e8a2e01c9f4ed", size = 11489751, upload-time = "2026-07-23T19:11:21.74Z" },
{ url = "https://files.pythonhosted.org/packages/b2/1d/1b2ef7bcde851c78d7f17f1cca13fd6dc695fc4b3d6197941e72cae5b132/ruff-0.16.0-py3-none-win32.whl", hash = "sha256:7fab76fa065c873f41ff744347c6e77bcc3dfec4bcc754dc26b63d23c0f7f5fb", size = 10785885, upload-time = "2026-07-23T19:11:23.947Z" },
{ url = "https://files.pythonhosted.org/packages/b2/a3/d5e4ef7a56be3f928ffb90b94c25ba7d3cb9c7fe0736aeaaedf361770712/ruff-0.16.0-py3-none-win_amd64.whl", hash = "sha256:429c117f022bf481fabd9d551e7a3952b24c65e6ef44337ea09d90bebef14472", size = 11923141, upload-time = "2026-07-23T19:11:26.409Z" },
{ url = "https://files.pythonhosted.org/packages/cb/9a/8415f2657cbe200f41a4531ccededf135505a92d4a012229121f885b26f9/ruff-0.16.0-py3-none-win_arm64.whl", hash = "sha256:14296fedcd2705c77ab8235439278bbb38f285cf7da5528b00b3e330c3d4872d", size = 11273407, upload-time = "2026-07-23T19:11:28.705Z" },
]
[[package]]
name = "secretstorage"
version = "3.5.0"
@@ -1859,6 +2056,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/fd/19/4b4e3e2ea5fa213ff4220e84450628fecde042b0961e7b4e6d845e555ade/xattr-1.3.0-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:1e6c216927b16fd4b72df655d5124b69b2a406cb3132b5231179021182f0f0d1", size = 19023, upload-time = "2025-10-13T22:16:34.395Z" },
]
[[package]]
name = "zipp"
version = "4.1.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/b9/d8/eab98a517c14134c0b2eb4e2387bc5f457334293ec5d2dd3857ec2966802/zipp-4.1.0.tar.gz", hash = "sha256:4cb57381f544315db7688e976e922a2b18cdb513d21cc194eb42232ba2a3e602", size = 26214, upload-time = "2026-05-18T20:08:57.967Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/3a/13/547360d81e6d88d58492968ffda9f9542854f11310ee556fef14260cc886/zipp-4.1.0-py3-none-any.whl", hash = "sha256:25ad4e16390cd314347dd8f1de67a2ac538ae658ed4ab9db16029c07c188e97f", size = 10238, upload-time = "2026-05-18T20:08:57.045Z" },
]
[[package]]
name = "zstandard"
version = "0.25.0"