Initial

Author: root

Dockerfile

@@ -0,0 +1,36 @@
+# Python app
+FROM python:3.12-slim-bookworm
+
+ENV POETRY_VERSION=2.1.3
+ENV PYTHONBUFFERED=1
+ENV POETRY_INSTALLER_MAX_WORKERS=1
+ENV POETRY_VIRTUALENVS_IN_PROJECT=false
+ENV POETRY_VIRTUALENVS_CREATE=false
+
+RUN apt update \
+    && apt install -y \
+      curl \
+      libffi-dev  \
+    && curl -sSL https://install.python-poetry.org | python - --version ${POETRY_VERSION} \
+    && apt remove -y --autoremove --purge curl libffi-dev \
+    && apt clean && rm -rf /var/lib/apt/lists/*
+
+ENV PATH="/root/.local/bin:$PATH"
+
+WORKDIR /code
+
+COPY ./pyproject.toml ./poetry.lock /code/
+RUN poetry install --no-interaction --no-root --only=main
+
+ARG BUILD_COMMIT_SHA
+ENV BUILD_COMMIT_SHA ${BUILD_COMMIT_SHA:-}
+
+#RUN if [ "${BUILD_COMMIT_SHA}" = "localdev" ]; then \
+#    poetry install --no-interaction --no-root --only=dev; \
+#    fi
+
+COPY . /code
+ENV PYTHONUNBUFFERED=0
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]
+

compose.yml

@@ -0,0 +1,19 @@
+services:
+  auth-proxy:
+    pull_policy: never
+    restart: unless-stopped
+    image: auth-proxy
+    build:
+      context: ./
+      args:
+        - BUILD_COMMIT_SHA=localdev
+    #volumes:
+    #- "./:/code/"
+    ports:
+      - "8456:80"
+
+    #command: [
+    #"uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80", "--reload",
+    #]
+    environment:
+      - AUTHENTIK_TOKEN

main.py

@@ -0,0 +1,126 @@
+import os
+from contextlib import asynccontextmanager
+from datetime import datetime
+from typing import Annotated, Any
+
+import httpx
+
+from fastapi import FastAPI, HTTPException
+from fastapi_utilities import repeat_every
+from pydantic import AliasPath, BaseModel, Field
+from pydantic_settings import BaseSettings
+
+
+def transform_card_number_to_unique(card_number: str):
+    """
+    For new, longer card ID format, we need to transform it to the old format
+    New card ID is returned in the exact byte order that's on card, not reversed while shorter format was reversed
+    to match the decimal representation printed on key fobs, so it can be just converted to hex and put in LDAP.
+    """
+    if len(card_number) == 8:
+        return card_number[4:6] + card_number[2:4] + card_number[0:2]
+    else:
+        return card_number
+
+
+def transform_card_number_to_mifare(card_number: str):
+    """
+    For new, longer card ID format, we need to transform it to the old format
+    New card ID is returned in the exact byte order that's on card, not reversed while shorter format was reversed
+    to match the decimal representation printed on key fobs, so it can be just converted to hex and put in LDAP.
+    """
+    if len(card_number) == 6:
+        return card_number[4:6] + card_number[2:4] + card_number[0:2] + "00"
+    else:
+        return card_number
+
+
+class User(BaseModel):
+    uid: Annotated[str, Field(validation_alias="username")]
+    mifare_card_ids: Annotated[list[str], Field(validation_alias=AliasPath("attributes", "mifareCardId"))] = []
+    unique_card_ids: Annotated[list[str], Field(validation_alias=AliasPath("attributes", "uniquecardId"))] = []
+    membership_expiration: Annotated[int, Field(validation_alias=AliasPath("attributes", "membershipExpirationTimestamp"))]
+
+
+class Settings(BaseSettings):
+    authentik_token: str = ...
+
+
+config = Settings()
+
+
+users: list[User] = []
+users_by_card: dict[str, User] = {}
+users_last_success_run: datetime | None
+users_last_failed_run: datetime | None
+users_last_failed_reason: Any
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    await fetch_users()
+    yield
+
+
+app = FastAPI(lifespan=lifespan)
+
+
+def auth():
+    return
+
+
+async def fetch() -> list[User]:
+    async with httpx.AsyncClient(
+            headers={"Authorization": f"Bearer {config.authentik_token}"},
+            timeout=15.0,
+        ) as client:
+        response = await client.get("https://auth.apps.hskrk.pl/api/v3/core/users/?attributes={\"membershipExpirationTimestamp__gt\": 1734998400}&page_size=200")
+        return [
+            User(**u)
+            for u in response.json()['results']
+        ]
+
+
+@repeat_every(seconds=300)
+async def fetch_users():
+    global users
+    global users_by_card
+    users = await fetch()
+    users_by_card = {
+        **{
+            mifare.lower(): user
+            for user in users for mifare in user.mifare_card_ids
+        },
+        **{
+            transform_card_number_to_unique(mifare).lower(): user
+            for user in users for mifare in user.mifare_card_ids
+        },
+        **{
+            unique.lower(): user
+            for user in users for unique in user.unique_card_ids
+        },
+        **{
+            transform_card_number_to_mifare(unique).lower(): user
+            for user in users for unique in user.unique_card_ids
+        },
+    }
+
+
+@app.get("/users/-/stats")
+async def get_user_stats():
+    return {
+        "users": {
+            "count": len(users),
+        },
+        "cards": {
+            "count": len(users_by_card),
+        },
+    }
+
+
+@app.get("/users/-/by-card/{card_id}")
+async def get_user_by_card(card_id: str):
+    user = users_by_card.get(card_id.lower())
+    if user is None:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return user