astroklient: warstwa pośrednia — pełne astro, bez generowania i administracji (4/5)
Testy / Testy warstwy logicznej (silnik) (pull_request) Successful in 10m19s
Testy / Testy warstwy prezentacji (dostęp do baz) (pull_request) Successful in 9m29s
Testy / Testy warstwy bazodanowej (ochrona baz) (pull_request) Successful in 9m26s
Testy / Testy astrodemo (pull_request) Failing after 0s
Testy / Testy astroklient (pull_request) Successful in 9m29s
Testy / Build obrazu silnika B (swisseph) (pull_request) Successful in 7s
Testy / Kontrola składni wszystkich warstw (pull_request) Successful in 5s
build / build (push) Successful in 19s
Testy / Testy warstwy logicznej (silnik) (push) Successful in 10m25s
Testy / Testy warstwy prezentacji (dostęp do baz) (push) Successful in 9m29s
Testy / Testy warstwy bazodanowej (ochrona baz) (push) Successful in 9m26s
Testy / Testy astrodemo (push) Failing after 0s
Testy / Testy astroklient (push) Successful in 9m29s
Testy / Build obrazu silnika B (swisseph) (push) Successful in 7s
Testy / Kontrola składni wszystkich warstw (push) Successful in 5s

Trzeci produkt drabiny: astrodemo (dwie funkcje) → astroklient → astrololo.
Po trzech poprzednich krokach jest cienki, bo jest ZŁOŻENIEM, a nie kopią:
własne main.py z ośmioma importami, a rdzeń — ekrany, szablony, zasoby — bierze
z warstwy prezentacji przy budowaniu obrazu. Jedno źródło, dwa produkty; inaczej
te same 2500 linii szablonów żyłyby w dwóch egzemplarzach i rozjechały się
w ciągu tygodni, po cichu.

CO MA: Horoskop, Interpretacje, Kalendarz, Synastria, Sygnifikatory, wgrywanie
plików. Wyszukiwarka miejsca i strefa czasowa zgodnie z ustaleniem.

CZEGO NIE MA I DLACZEGO NIE DA SIĘ WŁĄCZYĆ: plików usuniętych wg usun.txt nie ma
w obrazie. Nie istnieje uprawnienie, którym dałoby się je odsłonić, bo katalog
funkcji składa się ze ZGŁOSZEŃ ekranów obecnych w obrazie. To dlatego „każde
konto dostaje wszystko, co ta usługa umie" jest tu bezpieczne i nie wymaga
wypisywania listy: zbiór liczy się z katalogu, więc opisuje ten produkt.

KONTA jak w astrodemo: z konfiguracji środowiska (ASTROKLIENT_USERS), jeden
poziom dostępu, bez pliku kont i bez ekranu ich zakładania. Konta rozdziela się
po to, żeby każde miało własną pulę plików.

PULE PER KONTO — tu była realna dziura. Warstwa logiczna przenosiła pulę tylko
przy raporcie i operacjach na plikach, więc Kalendarz i Sygnifikatory czytałyby
CAŁY udział: jedno konto widziałoby pliki drugiego, mimo obietnicy izolacji.
Domknięte: TimelineRequest i QueryRequest niosą teraz pulę, a QueryService buduje
klienta danych na żądanie.

Pula jedzie w każdym żądaniu w dół i bierze się z kontekstu ustawianego przy
wejściu, nigdy z formularza. Test podstawia `tenant=ktos-inny` w POST i sprawdza,
że w dół poszedł login zalogowanego.

WARSTWA WSPÓLNA ROZDZIELONA OD POJĘCIA ADMINISTRATORA. base.html miał wpisany na
sztywno warunek `can(request, 'admin')` i odsyłacz do ekranu kont — czyli
w produkcie bez tego ekranu zostawał martwy link i nazwa czegoś, czego nie ma.
Rejestr niesie teraz wymagane uprawnienie, a szablon dostaje gotową listę.
Podstawa przestała też importować moduły służące jednemu ekranowi (konta, stany
plików), bo produkt bez tego ekranu wlókł ich zależności.

ZAPORA SŁOWNIKOWA NAD REALNYM DRZEWEM. Test buduje złożenie tak samo jak
Dockerfile i szuka słów o funkcjach, których nie ma — w odpowiedziach ORAZ
w plikach. Pierwsza wersja znalazła dziesięć trafień, w tym trzy moje własne
docstringi WYLICZAJĄCE nieobecne funkcje: zdanie „nie ma tu generowania tekstu"
mówi wprost, że coś takiego istnieje, więc jest takim samym śladem jak przycisk.
Po poprawkach: zero. Test ma kontrolę negatywną — podrzucony plik ma go wywrócić.

usun.txt jest DANYMI, nie tekstem w Dockerfile: czyta go też test pilnujący, żeby
zgadzał się ze złożeniem w main.py. Rozjazd znaczyłby albo martwy kod w obrazie,
albo błąd dopiero przy uruchomieniu.

CI: astroklient buduje się z KORZENIA repozytorium (jego Dockerfile sięga po
rdzeń), a zmiana w warstwie prezentacji też go przebudowuje — bez tego jego obraz
zostawałby ze starymi ekranami, a różnicy nie byłoby widać do zgłoszenia
użytkownika.

Testy: astroklient 12, presentation 368, astrodemo 28, logic 342, data 42,
render 41.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit was merged in pull request #83.
This commit is contained in:
2026-08-26 20:40:38 +02:00
parent cf24c4d473
commit 10970c579f
28 changed files with 1255 additions and 31 deletions
+18 -2
View File
@@ -24,7 +24,7 @@ jobs:
run: |
set -eu
TAG=${GITHUB_SHA::8}
USLUGI="data logic presentation astrodemo"
USLUGI="data logic presentation astrodemo astroklient"
# Usługi z własnym pipeline'em — celowo poza tą pętlą.
OSOBNE="render engine-swisseph"
@@ -67,6 +67,13 @@ jobs:
DO_BUDOWY="$DO_BUDOWY $SVC"
fi
done
# astroklient bierze rdzeń z warstwy prezentacji, więc zmiana TAMTEJ
# też go dotyczy. Bez tego jego obraz zostawałby ze starymi ekranami,
# a różnicy nie byłoby widać aż do zgłoszenia użytkownika.
if echo "$ZMIENIONE" | grep -q "^services/presentation/" \
&& ! echo "$DO_BUDOWY" | grep -q "astroklient"; then
DO_BUDOWY="$DO_BUDOWY astroklient"
fi
fi
if [ -z "$(echo "$DO_BUDOWY" | tr -d ' ')" ]; then
@@ -77,7 +84,16 @@ jobs:
echo "Buduję:$DO_BUDOWY (tag $TAG)"
for SVC in $DO_BUDOWY; do
docker build -t gitea.czernobog.pl/gitea/astrololo-$SVC:$TAG ./services/$SVC
# astroklient buduje się z KORZENIA repozytorium, bo jego Dockerfile
# sięga po rdzeń do services/presentation. Pozostałe mają kontekst
# ograniczony do własnego katalogu — i tak ma zostać, bo to właśnie
# ten kontekst gwarantuje, że nie wciągną niczego spoza siebie.
if [ "$SVC" = "astroklient" ]; then
docker build -f services/astroklient/Dockerfile \
-t gitea.czernobog.pl/gitea/astrololo-$SVC:$TAG .
else
docker build -t gitea.czernobog.pl/gitea/astrololo-$SVC:$TAG ./services/$SVC
fi
docker push gitea.czernobog.pl/gitea/astrololo-$SVC:$TAG
done
echo "Tag: $TAG"
+18
View File
@@ -108,6 +108,24 @@ jobs:
# żadna trasa poza dwiema funkcjami.
- name: Testy (pytest)
working-directory: services/astrodemo
astroklient-tests:
name: Testy astroklient
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
cache: pip
cache-dependency-path: services/astroklient/requirements-dev.txt
- name: Instalacja zależności
run: pip install -r services/astroklient/requirements-dev.txt
# Testy budują drzewo tak samo, jak robi to Dockerfile: rdzeń z warstwy
# prezentacji, minus usun.txt, plus nakładka. Sprawdzają WYNIK złożenia,
# bo tylko on trafia do obrazu.
- name: Testy (pytest)
working-directory: services/astroklient
env:
PYTHONPATH: .
run: pytest tests -q -rs
+28
View File
@@ -0,0 +1,28 @@
FROM python:3.12-slim
WORKDIR /app
COPY services/astroklient/requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Rdzeń wspólny z pełną aplikacją: ekrany astrologiczne, szablony, zasoby.
# Jedno źródło, dwa produkty — inaczej te same 2500 linii szablonów żyłyby
# w dwóch egzemplarzach i rozjechały się w ciągu tygodni, po cichu.
COPY services/presentation/app ./app
# …minus wszystko, czego ten produkt nie ma. Lista jest DANYMI, a nie tekstem
# w tym pliku: czyta ją także test, który pilnuje, żeby zgadzała się ze
# złożeniem w main.py. Rozjazd między nimi znaczyłby albo martwy kod w obrazie,
# albo brakujący plik i błąd dopiero przy uruchomieniu.
COPY services/astroklient/usun.txt .
RUN while read -r SCIEZKA; do \
[ -n "$SCIEZKA" ] || continue; \
test -e "app/$SCIEZKA" || { echo "usun.txt wskazuje nieistniejące: $SCIEZKA"; exit 1; }; \
rm -rf "app/$SCIEZKA"; \
done < usun.txt && rm usun.txt
# Nakładka: własne logowanie, własna pula plików, własny ekran wgrywania,
# własne złożenie. Nadpisuje pliki o tych samych nazwach.
COPY services/astroklient/app ./app
EXPOSE 8006
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8006"]
@@ -0,0 +1,174 @@
"""Klient warstwy logicznej — z PULĄ KONTA (PRE-29).
Każde konto ma własny, niewidoczny dla pozostałych zbiór plików. Pula jedzie
w każdym żądaniu w dół i bierze się WYŁĄCZNIE z kontekstu ustawianego przy
wejściu, nigdy z pola formularza: gdyby przychodziła z żądania, wystarczyłoby
podstawić cudzy login, żeby czytać cudze pliki.
Kontekst zamiast argumentu, bo ekrany są wspólne z pełną aplikacją i wołają
klienta jako obiekt modułowy. Przekazywanie puli przez wszystkie wywołania
znaczyłoby, że pominięcie jednego z nich cicho otwiera dostęp do całego udziału —
a takiego pominięcia nie widać.
"""
from __future__ import annotations
import contextvars
import os
from typing import Any
import httpx
# Login zalogowanego konta na czas obsługi żądania. Ustawia go bramka wejścia.
KONTO: contextvars.ContextVar[str] = contextvars.ContextVar("konto", default="")
from app import link_crypto
from app.config import settings
def _auth_headers() -> dict[str, str]:
"""Token międzywarstwowy (LOG-32) — pusty, gdy ochrona wyłączona."""
token = os.getenv("INTERNAL_TOKEN", "")
return {"X-Astrololo-Token": token} if token else {}
def _link() -> link_crypto.Link | None:
"""Klucz łącza prezentacja↔logika. Czytany przy każdym wywołaniu, bo
konfiguracja może się zmienić bez restartu procesu (testy, podmiana sekretu)."""
key = link_crypto.key_from_env(link_crypto.ENV_PRESENTATION_LOGIC)
return link_crypto.Link(key) if key else None
class LogicClient:
def __init__(self, base_url: str | None = None) -> None:
self.base_url = (base_url or settings.logic_url).rstrip("/")
def _post(self, path: str, payload: dict[str, Any], timeout: float) -> dict[str, Any]:
# Pula konta dokładana do KAŻDEGO żądania w dół. Bierze się z kontekstu
# ustawianego przy wejściu, nigdy z pola formularza — gdyby przychodziła
# z żądania, wystarczyłoby podstawić cudzy login, żeby czytać cudze pliki.
# Trasy, których pula nie dotyczy, po prostu ignorują nadmiarowe pole.
payload = {**payload, "tenant": KONTO.get("")}
"""Jedyna droga w dół. Celowo JEDNA: dopóki każda metoda składała żądanie
sama, dołożenie nowej znaczyło, że łatwo zapomnieć o tokenie albo kluczu
łącza — i tak się już raz stało (401 wyszedł dopiero na produkcji)."""
with httpx.Client(timeout=timeout) as client:
return link_crypto.call_json(client, "POST", f"{self.base_url}{path}",
payload=payload, headers=_auth_headers(),
link=_link())
# ── transport dla modułów opcjonalnych ─────────────────────────────────
# Moduł nie może dopisywać sobie metod do tej klasy: jego nazwy siedziałyby
# wtedy w pliku, który jedzie do KAŻDEGO obrazu. Dostaje więc samą drogę
# w dół, a co nią wysyła, jest jego sprawą. Szyfrowanie łącza i token
# międzywarstwowy obowiązują tak samo — to nie jest obejście ochrony.
def wywolaj(self, sciezka: str, payload: dict[str, Any], timeout: float) -> dict[str, Any]:
return self._post(sciezka, payload, timeout)
def pobierz(self, sciezka: str, timeout: float | None = None) -> dict[str, Any]:
with httpx.Client(timeout=timeout or settings.http_timeout) as client:
return link_crypto.call_json(client, "GET", f"{self.base_url}{sciezka}",
headers=_auth_headers(), link=_link())
def strumien(self, sciezka: str, payload: dict[str, Any]):
"""Strumień NDJSON w dół — przekazywany dalej ramka po ramce.
Idzie przez szyfrowane łącze jak reszta ruchu (PRE-16): bez tego przy
włączonym LINK_ENCRYPTION_REQUIRED serwer odrzuciłby żądanie."""
with httpx.Client(timeout=httpx.Timeout(None, connect=15.0)) as client:
yield from link_crypto.stream_lines(
client, f"{self.base_url}{sciezka}",
payload=payload, headers=_auth_headers(), link=_link())
def query(self, query: str, field: str, exact: bool, limit: int) -> dict[str, Any]:
payload = {"query": query, "field": field, "exact": exact, "limit": limit}
return self._post("/api/query", payload, settings.http_timeout)
def positions(
self,
when_utc_iso: str,
lat: float,
lon: float,
objects: list[str] | None = None,
house_system: str = "whole_sign",
stations: bool = False,
zodiac: str = "tropical",
tables: bool = False,
house_systems: list[str] | None = None,
aspect_orb: float = 8.0,
aspect_luminary_bonus: float = 2.0,
aspect_minor: bool = False,
) -> dict[str, Any]:
"""Pełny horoskop dla danego momentu — woła logic /chart/positions."""
payload = {
"when_utc": when_utc_iso,
"lat": lat,
"lon": lon,
"objects": objects,
"house_system": house_system,
"house_systems": house_systems,
"aspect_orb": aspect_orb,
"aspect_luminary_bonus": aspect_luminary_bonus,
"aspect_minor": aspect_minor,
"stations": stations,
"zodiac": zodiac,
"tables": tables,
}
# stacje ORAZ tabele wymagają root-findów / szukania numerycznego — dłuższy timeout
timeout = max(settings.http_timeout, 60.0) if (stations or tables) else settings.http_timeout
return self._post("/chart/positions", payload, timeout)
def synastry(
self, person_a: dict, person_b: dict, zodiac: str = "tropical",
aspect_orb: float = 8.0, aspect_luminary_bonus: float = 2.0, aspect_minor: bool = False,
) -> dict[str, Any]:
"""Synastria dwóch osób (PRE-04) — aspekty między horoskopami."""
payload = {
"person_a": person_a, "person_b": person_b, "zodiac": zodiac,
"aspect_orb": aspect_orb, "aspect_luminary_bonus": aspect_luminary_bonus,
"aspect_minor": aspect_minor,
}
return self._post("/chart/synastry", payload, settings.http_timeout)
def report(
self, when_utc_iso: str, lat: float, lon: float, limit: int = 5000, group: bool = False
) -> dict[str, Any]:
"""Sygnifikatory z obliczeń szukane w bazie — woła logic /chart/report."""
payload = {"when_utc": when_utc_iso, "lat": lat, "lon": lon, "limit": limit, "group": group}
return self._post("/chart/report", payload, max(settings.http_timeout, 30.0))
def bases(self) -> dict[str, Any]:
"""Bazy dostępne na udziale + metaopis i stan włączenia (DAN-15/PRE-09)."""
with httpx.Client(timeout=settings.http_timeout) as client:
return link_crypto.call_json(client, "GET", f"{self.base_url}/bases",
headers=_auth_headers(), link=_link())
# ── zarządzanie plikami baz (DAN-27) ────────────────────────────────
# Jedna metoda na trasę, bez sprytnego generyka: te wywołania różnią się
# skutkiem (odczyt / zapis / skasowanie), a ujednolicenie ich w jedno
# `call(path, payload)` zaciera tę różnicę dokładnie tam, gdzie jest ważna.
def files_status(self, path: str, status: str, by: str = "") -> dict[str, Any]:
return self._files_post("/files/status", {"path": path, "status": status, "by": by})
def files_upload(self, filename: str, content_b64: str, by: str = "") -> dict[str, Any]:
return self._files_post("/files/upload",
{"filename": filename, "content_b64": content_b64, "by": by})
def _files_post(self, path: str, payload: dict) -> dict[str, Any]:
payload = {**payload, "tenant": KONTO.get("")}
with httpx.Client(timeout=settings.http_timeout) as client:
return link_crypto.call_json(client, "POST", f"{self.base_url}{path}",
payload=payload, headers=_auth_headers(), link=_link())
def timeline(
self, when_utc_iso: str, lat: float, lon: float,
from_date: str, to_date: str, interpret: bool = True,
) -> dict[str, Any]:
"""Oś czasu z technik (+interpretacje z bazy) — woła logic /chart/timeline."""
payload = {
"when_utc": when_utc_iso, "lat": lat, "lon": lon,
"from_date": from_date, "to_date": to_date, "interpret": interpret,
}
return self._post("/chart/timeline", payload, max(settings.http_timeout, 60.0))
@@ -0,0 +1,75 @@
"""Wgrywanie plików: jedna czynność, nie dwie.
„Dodać plik" ma tu znaczyć, że od razu bierze udział w wyszukiwaniu — inaczej po
wgraniu nic by się nie zmieniło i wyglądałoby to na zepsute. Nie ma listy plików
ani ich stanów: ta usługa ma jeden poziom dostępu, więc nie ma komu ich
pokazywać ani po co.
"""
from __future__ import annotations
import base64
import httpx
from fastapi import File, Request, UploadFile
from fastapi.responses import HTMLResponse, RedirectResponse
from app.features import Feature
from app.podstawa import app, features, logic, templates, _logic_error
# Stan pliku „w użyciu" — ta sama nazwa, co w warstwie danych (DAN-27). Wpisana
# tutaj wprost, bo ta usługa tamtej warstwy nie importuje; literówka objawiłaby
# się dopiero na żywo, więc pilnuje jej test.
W_UZYCIU = "active"
@app.get("/pliki", response_class=HTMLResponse)
def wgrywanie_form(request: Request, error: str = "", done: str = ""):
return templates.TemplateResponse(request, "wgrywanie.html",
{"error": error, "done": done})
@app.post("/pliki/wgraj")
async def wgraj(request: Request, upload: UploadFile = File(...)):
"""Wgranie ORAZ włączenie do użytku.
Plik o złym układzie nie wchodzi do użytku, ale NIE JEST tracony — zostaje
zapisany i czeka. Komunikat nie mówi, CZYM się nie nadaje: różnica w treści
byłaby wyrocznią, z której da się odczytać, czego program oczekuje.
"""
raw = await upload.read()
if not raw:
return _wroc(error="Pusty plik.")
kto = getattr(request.state, "user", "") or "?"
try:
out = logic.files_upload(upload.filename or "dane.xlsx",
base64.b64encode(raw).decode("ascii"), by=kto)
except httpx.HTTPError as e:
return _wroc(error=_logic_error(e))
nazwa = out.get("name") or "plik"
if not out.get("accepted"):
return _wroc(done=f"Plik „{nazwa}” został przyjęty, ale nie nadaje się "
f"do użycia w tej postaci.")
try:
logic.files_status(out["path"], W_UZYCIU, by=kto)
except httpx.HTTPError as e:
return _wroc(error=f"Plik „{nazwa}” wgrany, ale nie udało się go włączyć.")
return _wroc(done=f"Dodano „{nazwa}”. Plik jest już używany przy wyszukiwaniu.")
def _wroc(error: str = "", done: str = "") -> RedirectResponse:
"""Po zapisie PRZEKIEROWANIE, nie render — odświeżenie strony nie może
wgrać pliku drugi raz."""
from urllib.parse import urlencode
q = urlencode({k: v for k, v in (("error", error), ("done", done)) if v})
return RedirectResponse(f"/pliki{'?' + q if q else ''}", status_code=303)
features.zarejestruj(
ekran=Feature("files", "Pliki", "Dodawanie plików używanych przy wyszukiwaniu.",
"Ekrany", "/pliki", kolejnosc=70),
trasy={("GET", "/pliki"): "files", ("POST", "/pliki/wgraj"): "files"},
zasoby={"s-wgrywanie.css": "files"},
)
+34
View File
@@ -0,0 +1,34 @@
"""Astroklient — ZŁOŻENIE produktu.
Pełne funkcje astrologiczne, bez generowania tekstu i bez administracji. Ten plik
nie zawiera żadnej trasy: odpowiada wyłącznie na pytanie, z czego składa się ten
produkt. Import ekranu rejestruje jego trasy, więc lista poniżej JEST definicją.
Poza tą listą nie ma nic. Nie chodzi o funkcje wyłączone: plików, których tu nie
widać, nie ma w obrazie (patrz usun.txt), więc nie istnieje uprawnienie, którym
dałoby się cokolwiek odsłonić — ani nazwa, po której dałoby się zgadnąć, czego
szukać.
"""
from __future__ import annotations
from app import rozszerzenia
from app.podstawa import (_asset_body, _asset_version, _build_utc, _logic_error, app,
logic, static_url, templates)
from app.ekrany import ( # noqa: F401 - import rejestruje trasy
horoskop,
interpretacje,
kalendarz,
logowanie,
lokalizacja,
sygnifikatory,
synastria,
wgrywanie,
)
__all__ = ["app", "logic", "templates", "static_url", "_asset_body", "_asset_version"]
# Most modułów opcjonalnych zostaje CELOWO. W tym obrazie nie ma katalogu
# `app/moduly/`, więc odkrywa zero modułów i wszystkie odpowiedzi są puste —
# aplikacja zachowuje się tak, jakby takiego pojęcia nigdy nie było.
rozszerzenia.zamontuj(app, logic=logic, templates=templates,
buduj_utc=_build_utc, blad=_logic_error)
+384
View File
@@ -0,0 +1,384 @@
"""Logowanie i bramka uprawnień astroklienta.
Konta pochodzą WYŁĄCZNIE z konfiguracji środowiska (`ASTROKLIENT_USERS`), a każde
dostaje ten sam zestaw: wszystko, co ta usługa umie. Konta rozdziela się po to,
żeby każde miało własną, niewidoczną dla pozostałych pulę plików — nie po to, żeby
dawać im różne poziomy dostępu. Poziom jest jeden i nie ma go z czym porównywać.
Reszta jest wspólna z pozostałymi usługami i celowo nietknięta: hasła jako scrypt,
sesja w podpisanym ciasteczku, limit żądań, dziennik audytowy, oraz odmowa jako
404 zamiast 403 — bo 403 samo w sobie mówi, że coś tam jest.
"""
from __future__ import annotations
import base64
import binascii
import hashlib
import hmac
import logging
import os
import secrets
import time
from collections import deque
from dataclasses import dataclass
from fastapi import Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
log = logging.getLogger("astrololo.security")
MAX_TRACKED_CLIENTS_DEFAULT = 4096
MAX_TRACKED_CLIENTS = MAX_TRACKED_CLIENTS_DEFAULT # zabezpieczenie przed puchnięciem pamięci
# ── konta ─────────────────────────────────────────────────────────────────
# Wyłącznie z konfiguracji środowiska. Konta rozdziela się po to, żeby każde
# miało własną pulę plików — nie po to, żeby różniły się tym, co widzą.
# ── konta imienne (PRE-17) ───────────────────────────────────────────────
# Parametry scrypt: rozsądny kompromis czas/pamięć dla logowania do aplikacji
# wewnętrznej. Zmiana wymaga przeliczenia haseł (hash niesie tylko sól).
_SCRYPT_N, _SCRYPT_R, _SCRYPT_P, _SCRYPT_LEN = 2**14, 8, 1, 32
_HASH_PREFIX = "scrypt$"
def hash_password(password: str, salt: bytes | None = None) -> str:
"""Hasło → wpis do ASTROKLIENT_USERS: `scrypt$<sól_hex>$<hash_hex>`."""
salt = salt or secrets.token_bytes(16)
dk = hashlib.scrypt(password.encode("utf-8"), salt=salt,
n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P, dklen=_SCRYPT_LEN)
return f"{_HASH_PREFIX}{salt.hex()}${dk.hex()}"
def accounts() -> dict[str, str]:
"""Konta z `ASTROKLIENT_USERS` — `login:sekret`, po przecinku.
Sekret to hash scrypt (`scrypt$sól$hash`), ten sam format co w pozostałych
usługach — jedna implementacja, więc nie ma czego rozjechać."""
out: dict[str, str] = {}
for entry in os.getenv("ASTROKLIENT_USERS", "").split(","):
entry = entry.strip()
if not entry:
continue
user, sep, secret = entry.partition(":")
if sep and user.strip() and secret.strip():
out[user.strip()] = secret.strip()
return out
def _secret_matches(spec: str, password: str) -> bool:
"""Sekret konta vs podane hasło. Hash scrypt albo (zgodność wstecz) jawne."""
if spec.startswith(_HASH_PREFIX):
try:
_, salt_hex, hash_hex = spec.split("$", 2)
salt, expected = bytes.fromhex(salt_hex), bytes.fromhex(hash_hex)
except ValueError:
return False
actual = hashlib.scrypt(password.encode("utf-8"), salt=salt,
n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P, dklen=_SCRYPT_LEN)
return hmac.compare_digest(actual, expected)
# Porównujemy BAJTY, nie tekst: `compare_digest` na stringach rzuca TypeError
# przy znakach spoza ASCII, więc hasło z polskimi literami wywracało logowanie
# błędem 500 zamiast zwykłej odmowy (błąd zastany, złapany testem).
return hmac.compare_digest(spec.encode("utf-8"), password.encode("utf-8"))
def rate_limit_per_min() -> int:
return int(os.getenv("RATE_LIMIT_PER_MIN", "120"))
def cookies_secure() -> bool:
"""Czy ciasteczko sesji ma być wyłącznie dla https.
Domyślnie TAK. Wyłączamy jedynie w pracy lokalnej po http — inaczej
przeglądarka po prostu nie zapamięta ciasteczka i logowanie „nie działa"
bez żadnego komunikatu, co jest wyjątkowo mylącą awarią."""
return os.getenv("COOKIES_INSECURE", "").strip().lower() not in {"1", "true", "yes", "on"}
def trust_proxy() -> bool:
return os.getenv("TRUST_PROXY", "").strip().lower() in {"1", "true", "yes", "on"}
# /wyloguj jest „publiczna" tylko technicznie: przechodzi obok bramki, żeby móc
# odpowiedzieć 401 CELOWO, zamiast dostać 200 od zalogowanej sesji. Bez tego
# wylogowanie byłoby niemożliwe — przeglądarka dosyła zapamiętane dane przy każdym
# żądaniu, więc handler i tak widziałby zalogowaną osobę.
# /logowanie MUSI stać poza bramką — inaczej żeby się zalogować, trzeba by być
# zalogowanym. /wyloguj też: ma działać także wtedy, gdy sesja już wygasła.
# Ekran logowania musi się wyświetlić PRZED zalogowaniem, więc jego arkusz jest
# jedynym publicznym zasobem — i dlatego nie wolno w nim trzymać niczego, co
# nazywa funkcję (patrz features.STATIC).
PUBLIC_PATHS = frozenset({"/health", "/logowanie", "/wyloguj", "/static/base.css"})
# Pusto CELOWO. Dopóki był tu "/static/", cały katalog omijał bramkę: każdy plik
# JS i arkusz stylów pobierał ktokolwiek, także niezalogowany, pod zgadywalnym
# adresem — a ich treść wymienia ekrany, dostawców modeli i przeznaczenie baz.
# Ruch do zasobów nie trafiał przy tym ani do dziennika, ani pod limit żądań,
# więc wyciek był dla właściciela niewidoczny.
PUBLIC_PREFIXES: tuple[str, ...] = ()
_hits: dict[str, deque[float]] = {}
def auth_enabled() -> bool:
return bool(accounts())
def _is_public(path: str) -> bool:
return path in PUBLIC_PATHS or path.startswith(PUBLIC_PREFIXES)
@dataclass(frozen=True)
class Principal:
"""Kto się zalogował i co wolno mu zobaczyć."""
login: str
permissions: frozenset[str]
is_admin: bool
def _uprawnienia() -> frozenset[str]:
"""Każde konto dostaje wszystko, co ta usługa umie — liczone z KATALOGU.
Katalog składa się ze zgłoszeń ekranów obecnych w obrazie, więc ten zbiór
opisuje dokładnie ten produkt, a nie listę wpisaną kiedyś ręcznie. Dopisanie
ekranu nie wymaga pamiętania o niczym; usunięcie go zabiera uprawnienie samo.
Nie ma tu uprawnienia administracyjnego i nie ma go skąd wziąć: `features.ADMIN`
celowo nie należy do GRANTABLE."""
from app import features
return frozenset(features.GRANTABLE)
def _sekret(login: str) -> str | None:
return accounts().get(login)
def verify(login: str, password: str) -> Principal | None:
"""Sprawdza login i hasło. None = odmowa, bez podawania powodu.
Nieznany login i złe hasło dają ten sam wynik: rozróżnienie mówiłoby, które
konta istnieją, a to informacja warta czegoś dla zgadującego."""
spec = _sekret(login)
if spec is None or not _secret_matches(spec, password):
return None
return Principal(login, _uprawnienia(), False)
def issue_session(login: str) -> str:
"""Zawartość ciasteczka dla właśnie zalogowanego konta.
W ciasteczku ląduje ODCISK poświadczenia, dzięki czemu zmiana hasła albo
skasowanie konta unieważnia jego sesje same z siebie — bez pamiętania o tym
i bez żadnego magazynu sesji."""
from app import session
# Wersja sesji zawsze 0: nie ma trwałego magazynu, więc nie ma gdzie trzymać
# licznika pokolenia. Wylogowanie i zmiana hasła i tak działają — pierwsze
# przez skasowanie ciasteczka, drugie przez odcisk, który przestaje pasować.
return session.issue(login, session.fingerprint(_sekret(login) or ""), 0)
def principal(request: Request) -> Principal | None:
"""Ciasteczko → kto to jest. None = nie wpuszczać."""
from app import session
dane = session.read(request.cookies.get(session.COOKIE))
if dane is None:
return None
spec = _sekret(dane["login"])
if spec is None:
return None
# Zmiana hasła w konfiguracji ubija sesje tego konta — to jest tutaj jedyny
# sposób na „wyloguj mnie ze wszystkich urządzeń", bo bez trwałego magazynu
# nie ma gdzie trzymać licznika pokolenia sesji.
if session.fingerprint(spec) != dane["odcisk"]:
return None
return Principal(dane["login"], _uprawnienia(), False)
def client_ip(request: Request) -> str:
"""Adres, po którym rozliczamy limit żądań.
Za odwrotnym proxy (u nas: Ingress/Traefik po włączeniu TLS — PRE-16)
`request.client.host` to adres POD-a proxy, jednakowy dla wszystkich. Bez
poprawki cały ruch trafiałby do jednego wiadra i pierwsza osoba, która
wyklika limit, odcięłaby pozostałe.
Nagłówkom wierzymy WYŁĄCZNIE przy TRUST_PROXY — bo inaczej wystarczyłoby
dopisać własny `X-Forwarded-For`, żeby przy każdym żądaniu wyglądać na kogoś
innego i ominąć limit całkowicie. Z tego samego powodu bierzemy OSTATNI wpis
listy: to jedyny, który dopisał nasz proxy. Wcześniejsze mógł podstawić
klient, więc nie znaczą nic.
"""
peer = request.client.host if request.client else "?"
if not trust_proxy():
return peer
forwarded = request.headers.get("x-forwarded-for", "")
if forwarded:
last = forwarded.rsplit(",", 1)[-1].strip()
if last:
return last
return request.headers.get("x-real-ip", "").strip() or peer
def _rate_limited(client: str) -> bool:
cap = rate_limit_per_min()
if cap <= 0:
return False
now = time.monotonic()
window = _hits.get(client)
if window is None:
if len(_hits) >= MAX_TRACKED_CLIENTS:
_hits.clear() # prosty reset zamiast nieograniczonego wzrostu
window = _hits[client] = deque()
while window and now - window[0] > 60.0:
window.popleft()
if len(window) >= cap:
return True
window.append(now)
return False
# ── dziennik audytowy (PRE-17) ───────────────────────────────────────────
audit_log = logging.getLogger("astrololo.audit")
def _setup_audit_logging() -> None:
"""Gwarantuje, że wpisy audytowe FAKTYCZNIE wychodzą na stdout.
Domyślna konfiguracja uvicorna nie obsługuje naszych loggerów, więc wpisy na
poziomie INFO ginęły — dziennik istniał w kodzie, ale był pusty (sprawdzone
na żywym serwerze). Niewidoczny dziennik jest gorszy niż jego brak: daje
złudzenie kontroli. Własny handler na stdout, bo w k8s to stamtąd zbierane
są logi. `propagate=False` — żeby wpis nie dublował się przez korzeń.
"""
audit_log.setLevel(os.getenv("AUDIT_LEVEL", "INFO").upper())
if not audit_log.handlers:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
audit_log.addHandler(handler)
audit_log.propagate = False
def audit_records(request: Request, count: int) -> None:
"""Handler zgłasza, ILE rekordów baz oddał — trafi do wpisu audytowego.
Sama liczba, nigdy treść. Bez tego dziennik pokazywałby tylko „ktoś wchodził",
a nie „ktoś wyniósł 5000 rekordów" — a to drugie jest sygnałem nadużycia."""
try:
request.state.audit_records = int(count)
except (TypeError, ValueError):
pass
def audit_login(request: Request, login: str, udane: bool) -> None:
"""Zdarzenie logowania w dzienniku (PRE-17).
Nieudane próby są tu ważniejsze od udanych: pojedyncza nic nie znaczy, ale
seria pod jednym adresem to jedyny widoczny ślad zgadywania haseł. Logujemy
LOGIN, nigdy hasła ani jego fragmentu — dziennik to kolejny nośnik wycieku."""
audit_log.info("AUDYT zdarzenie=%s user=%s ip=%s",
"logowanie-ok" if udane else "logowanie-odmowa",
login or "-", client_ip(request))
def audit_logout(request: Request) -> None:
audit_log.info("AUDYT zdarzenie=wylogowanie user=%s ip=%s",
getattr(request.state, "user", "") or "-", client_ip(request))
def _audit(request: Request, user: str, status: int, ms: float) -> None:
records = getattr(request.state, "audit_records", None)
audit_log.info(
"AUDYT user=%s ip=%s method=%s path=%s status=%s records=%s ms=%.0f",
user or "-", client_ip(request), request.method, request.url.path,
status, "-" if records is None else records, ms,
)
def install(app) -> None:
"""Podpina ochronę pod wszystkie ścieżki poza publicznymi."""
from app import session
_setup_audit_logging()
# FAIL-CLOSED. Usługa z kontami, ale bez klucza podpisu, nie umiałaby odróżnić
# ważnej sesji od podrobionej. Lepiej, żeby nie wstała, niż żeby wpuszczała —
# i lepiej teraz, przy starcie, niż przy pierwszym logowaniu człowieka.
if auth_enabled() and not session.available():
raise session.SessionError(
"Konta są skonfigurowane, ale brakuje SESSION_SECRET — bez klucza "
"podpisu sesje nie działają, a logowanie nie miałoby czym się bronić. "
"Wygeneruj: openssl rand -hex 32")
if not auth_enabled():
log.warning(
"UWAGA: brak kont (ASTROKLIENT_USERS) — usługa jest OTWARTA dla "
"każdego, kto ma do niej dostęp sieciowy, wraz z treścią plików."
)
@app.middleware("http")
async def _guard(request: Request, call_next):
if _is_public(request.url.path):
return await call_next(request)
started = time.monotonic()
if _rate_limited(client_ip(request)):
_audit(request, "-", 429, (time.monotonic() - started) * 1000)
return JSONResponse(
{"detail": "Zbyt wiele żądań — spróbuj za chwilę."},
status_code=429, headers={"Retry-After": "60"},
)
from app import features
user, who, odswiez = "", None, None
if auth_enabled():
who = principal(request)
if who is None:
_audit(request, "-", 303, (time.monotonic() - started) * 1000)
# Przekierowanie na ekran logowania, nie 401: 401 z Basic otwierało
# okienko systemowe, którego nie da się ani opisać, ani z niego wyjść.
# Zapamiętujemy, dokąd ktoś zmierzał — po zalogowaniu wraca na miejsce.
from urllib.parse import quote
cel = request.url.path
dokad = f"?dokad={quote(cel, safe='/')}" if request.method == "GET" and cel != "/" else ""
return RedirectResponse(f"/logowanie{dokad}", status_code=303)
user = who.login
# Ruch przedłuża sesję, ale ciasteczko odsyłamy tylko wtedy, gdy
# znacznik zdążył się zestarzeć — inaczej Set-Cookie leciałby przy
# każdym obrazku i arkuszu stylów.
from app import session as _s
dane = _s.read(request.cookies.get(_s.COOKIE))
odswiez = _s.refresh(dane) if dane else None
else:
# Bez kont ochrona jest wyłączona (dev), więc widać wszystko — inaczej
# deweloper zobaczyłby pustą aplikację i szukałby błędu tam, gdzie go nie ma.
who = Principal("", _uprawnienia(), False)
request.state.user = user
request.state.principal = who
# Pula plików na czas obsługi tego żądania. Ustawiana TU, w jednym
# miejscu, zaraz po ustaleniu kto to jest — klient czyta ją z kontekstu,
# więc żadne wywołanie w dół nie może o niej zapomnieć.
from app.clients.logic_client import KONTO
KONTO.set(user)
# 404, NIE 403. Odmowa z powodem jest informacją, że coś tam jest —
# a konto ograniczone ma widzieć program kompletny, tylko mniejszy.
need = features.required(request.method, request.url.path)
if not features.can(who.permissions, need):
_audit(request, user, 404, (time.monotonic() - started) * 1000)
return HTMLResponse("<h1>404 — nie znaleziono</h1>", status_code=404)
response = await call_next(request)
if odswiez:
from app import session as _s
response.set_cookie(_s.COOKIE, odswiez, max_age=_s.max_age(),
**_s.cookie_params(secure=cookies_secure()))
_audit(request, user, response.status_code, (time.monotonic() - started) * 1000)
return response
+186
View File
@@ -0,0 +1,186 @@
"""Sesje logowania — podpisane ciasteczko (LOG-34).
CO ZASTĘPUJE I DLACZEGO. HTTP Basic nie ma wylogowania: przeglądarka zapamiętuje
hasło i dosyła je SAMA przy każdym żądaniu, więc serwer nie ma czego zapomnieć.
Sesja odwraca tę zależność — to serwer decyduje, czy dana przeglądarka jest
w środku, i może tę decyzję cofnąć.
TRZY POZIOMY UNIEWAŻNIENIA, celowo rozdzielone, bo każdy kosztuje co innego:
1. WYLOGOWANIE = skasowanie ciasteczka. Natychmiastowe, bez żadnego magazynu,
działa wszędzie. To jest to, czego oczekuje człowiek klikający „Wyloguj".
2. ZMIANA POŚWIADCZEŃ = odcisk hasła wpisany w ciasteczko przestaje pasować.
Zmiana hasła albo skasowanie konta ubija jego sesje SAMO, bez pamiętania
o tym — i też nie wymaga magazynu.
3. WYLOGOWANIE ZDALNE („wyrzuć ze wszystkich urządzeń") = licznik wersji,
który musi gdzieś leżeć. Wymaga zapisu, więc jest opcjonalny: gdy aplikacja
nie poda licznika, poziomy 1 i 2 nadal działają.
Poziom 2 jest tu najważniejszy: bez niego odebranie komuś dostępu przez zmianę
hasła zostawiałoby jego otwartą sesję żywą aż do wygaśnięcia — czyli odebranie
dostępu nie odbierałoby dostępu.
KLUCZ JEST WYMAGANY i bez niego usługa się nie uruchamia. Losowanie przy starcie
byłoby wygodne, ale oznaczałoby wylogowanie wszystkich przy każdym restarcie
poda — co wygląda jak awaria i uczy ludzi ignorować ekran logowania.
Podpis: HMAC-SHA256 ze stdlib. Bez nowych zależności — ta sama linia co scrypt
w PRE-27 i AES w PRE-16.
"""
from __future__ import annotations
import base64
import hashlib
import hmac
import json
import os
import time
# Nazwa własna tej usługi. Gdyby dwie aplikacje pod jedną domeną używały
# ciasteczka o tej samej nazwie, nadpisywałyby się nawzajem i człowiek wypadałby
# z jednej, logując się do drugiej.
COOKIE = "astroklient_sesja"
# Twardy kres życia sesji i kres bezczynności. Oba w sekundach, oba z konfiguracji.
DEFAULT_MAX_AGE = 30 * 24 * 3600 # 30 dni — po tylu i tak każe się zalogować
DEFAULT_IDLE = 8 * 3600 # 8 godzin bez ruchu = koniec
# Co ile odświeżamy znacznik aktywności w ciasteczku. Bez tego progu każde
# żądanie odsyłałoby Set-Cookie, także obrazki i arkusze stylów.
REFRESH_AFTER = 300
class SessionError(RuntimeError):
"""Sesje nie mogą działać — brak klucza albo klucz nie do przyjęcia."""
def _env_int(name: str, default: int) -> int:
try:
return int(os.getenv(name, "") or default)
except ValueError:
return default
def max_age() -> int:
return _env_int("SESSION_MAX_AGE", DEFAULT_MAX_AGE)
def idle_timeout() -> int:
return _env_int("SESSION_IDLE", DEFAULT_IDLE)
def secret() -> bytes:
"""Klucz podpisu. Brak = wyjątek, NIE cichy losowy klucz.
Fail-closed jest tu celowe: usługa bez klucza nie umie odróżnić ważnej sesji
od podrobionej, więc lepiej, żeby nie wstała, niż żeby wpuszczała."""
raw = os.getenv("SESSION_SECRET", "").strip()
if not raw:
raise SessionError(
"Brak SESSION_SECRET — bez klucza podpisu sesje nie mogą działać. "
"Wygeneruj: openssl rand -hex 32")
if len(raw) < 32:
raise SessionError(
f"SESSION_SECRET ma {len(raw)} znaków — za krótki. Wymagane co najmniej 32; "
f"krótki klucz da się odgadnąć, a wtedy podrobienie sesji jest trywialne.")
return raw.encode("utf-8")
def available() -> bool:
"""Czy sesje da się w ogóle obsłużyć (do sprawdzenia przy starcie)."""
try:
secret()
except SessionError:
return False
return True
def fingerprint(secret_spec: str) -> str:
"""Odcisk poświadczenia konta — 16 znaków hasza z hasła (albo jego hasza).
NIE JEST to hasło ani nic, z czego da się je odtworzyć: bierzemy skrót
zapisanego sekretu, a ten dla kont zarządzanych sam jest już haszem scrypt.
Do ciasteczka trafia po to, żeby zmiana hasła albo skasowanie konta
unieważniły jego sesje same z siebie."""
return hashlib.sha256(("odcisk:" + (secret_spec or "")).encode("utf-8")).hexdigest()[:16]
def _sign(payload: bytes) -> bytes:
return hmac.new(secret(), payload, hashlib.sha256).digest()
def _b64(raw: bytes) -> str:
return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
def _unb64(text: str) -> bytes:
return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))
def issue(login: str, odcisk: str, wersja: int = 0, *, now: float | None = None) -> str:
"""Nowe ciasteczko sesji dla konta."""
teraz = int(now if now is not None else time.time())
payload = json.dumps(
{"u": login, "f": odcisk, "v": int(wersja), "wyd": teraz, "akt": teraz},
separators=(",", ":"), sort_keys=True).encode("utf-8")
return f"{_b64(payload)}.{_b64(_sign(payload))}"
def read(token: str | None, *, now: float | None = None) -> dict | None:
"""Ciasteczko → zawartość albo None, gdy cokolwiek się nie zgadza.
None znaczy „nie wpuszczać" i nie mówi dlaczego — powód jest tu bez wartości
dla wołającego, a rozróżnianie „zła sygnatura" od „wygasło" tylko pomaga
zgadywać. Wygaśnięcie i tak kończy się tym samym ekranem logowania."""
if not token or "." not in token:
return None
surowy, _, podpis = token.rpartition(".")
try:
payload = _unb64(surowy)
oczekiwany = _sign(payload)
podany = _unb64(podpis)
except (ValueError, SessionError):
return None
# Porównanie w stałym czasie: zwykłe == przecieka informację o tym, ile
# pierwszych bajtów podpisu się zgadza, a to wystarcza do podrobienia.
if not hmac.compare_digest(oczekiwany, podany):
return None
try:
dane = json.loads(payload.decode("utf-8"))
login, odcisk = str(dane["u"]), str(dane["f"])
wersja, wyd, akt = int(dane["v"]), int(dane["wyd"]), int(dane["akt"])
except (ValueError, KeyError, TypeError):
return None
teraz = int(now if now is not None else time.time())
# Ciasteczko z przyszłości oznacza przestawiony zegar albo podrobienie —
# w obu wypadkach nie ma powodu mu ufać.
if wyd > teraz + 60:
return None
if teraz - wyd > max_age() or teraz - akt > idle_timeout():
return None
return {"login": login, "odcisk": odcisk, "wersja": wersja, "wydano": wyd, "aktywnosc": akt}
def refresh(dane: dict, *, now: float | None = None) -> str | None:
"""Odświeżone ciasteczko, gdy znacznik aktywności zdążył się zestarzeć.
None = nie ma potrzeby nic odsyłać. Bez tego progu każde żądanie zwracałoby
Set-Cookie — także po arkusze stylów i obrazki."""
teraz = int(now if now is not None else time.time())
if teraz - dane["aktywnosc"] < REFRESH_AFTER:
return None
payload = json.dumps(
{"u": dane["login"], "f": dane["odcisk"], "v": dane["wersja"],
"wyd": dane["wydano"], "akt": teraz},
separators=(",", ":"), sort_keys=True).encode("utf-8")
return f"{_b64(payload)}.{_b64(_sign(payload))}"
def cookie_params(secure: bool = True) -> dict:
"""Parametry ciasteczka wspólne dla ustawiania i kasowania.
httponly — JavaScript nie ma po co go czytać, a bez tego jeden wstrzyknięty
skrypt wynosi sesję. samesite=strict — odcina większość ataków polegających
na tym, że obca strona wysyła żądanie w imieniu zalogowanego."""
return {"httponly": True, "samesite": "strict", "secure": secure, "path": "/"}
@@ -0,0 +1,4 @@
/* Arkusz ekranu „files” — dokładany tylko dla konta z tym uprawnieniem. */
.card { background: var(--panel); border: 1px solid var(--line); border-radius: 12px;
padding: 1rem 1.1rem; margin-top: .4rem; }
@@ -0,0 +1,23 @@
{% extends "base.html" %}
{% set nav_active = "files" %}
{% block title %}Pliki{% endblock %}
{% block styles %}<link rel="stylesheet" href="{{ static('s-wgrywanie.css') }}">
{% endblock %}
{% block content %}
{# Opis czysto operacyjny: co robi wgranie, i nic o tym, do CZEGO te dane służą
ani co się z nimi dzieje dalej. #}
<p class="muted">Wgrany plik jest od razu używany przy wyszukiwaniu.</p>
{% if error %}<p class="notice-warn">{{ error }}</p>{% endif %}
{% if done %}<p class="done-note">{{ done }}</p>{% endif %}
<div class="meta">Wgraj plik</div>
<form method="post" action="/pliki/wgraj" enctype="multipart/form-data" class="card">
<div class="grid">
<label>Plik <input type="file" name="upload" required></label>
</div>
<div class="actions"><button type="submit">Wgraj</button></div>
</form>
{% endblock %}
@@ -0,0 +1,2 @@
-r requirements.txt
pytest>=8.0
+14
View File
@@ -0,0 +1,14 @@
fastapi>=0.115
uvicorn[standard]>=0.34
httpx>=0.28
jinja2>=3.1
python-multipart>=0.0.20
# Szyfrowanie łącza między warstwami (PRE-16): AES-256-GCM + HKDF
cryptography>=44.0
# Strefa czasowa z lokalizacji (PRE-03): współrzędne → strefa IANA (offline, lekki
# wheel Rust), a offset/DST liczy stdlib zoneinfo. tzdata na wypadek slim-obrazu
# bez systemowej bazy stref.
tzfpy>=0.15
tzdata>=2024.1
# Eksport wyników do Excela — „tabela robocza" (DAN-23/PRE-10)
openpyxl>=3.1
+37
View File
@@ -0,0 +1,37 @@
"""Drzewo astroklienta budowane tak samo, jak buduje je Dockerfile.
Ta usługa nie ma własnego kompletu plików: bierze rdzeń z warstwy prezentacji,
usuwa z niego to, czego ten produkt nie ma (usun.txt), i nakłada własne. Testy
muszą patrzeć na WYNIK tego złożenia, a nie na katalog `app/` w repozytorium —
inaczej sprawdzałyby garść plików, a nie produkt.
"""
from __future__ import annotations
import pathlib
import shutil
import pytest
USLUGA = pathlib.Path(__file__).resolve().parents[1]
RDZEN = USLUGA.parents[0] / "presentation" / "app"
def zloz(cel: pathlib.Path) -> pathlib.Path:
"""Rdzeń minus usun.txt plus nakładka — dokładnie jak w Dockerfile."""
app = cel / "app"
shutil.copytree(RDZEN, app, ignore=shutil.ignore_patterns("__pycache__", "*.pyc"))
for linia in (USLUGA / "usun.txt").read_text(encoding="utf-8").splitlines():
sciezka = linia.strip()
if not sciezka:
continue
do_usuniecia = app / sciezka
assert do_usuniecia.exists(), f"usun.txt wskazuje nieistniejące: {sciezka}"
shutil.rmtree(do_usuniecia) if do_usuniecia.is_dir() else do_usuniecia.unlink()
shutil.copytree(USLUGA / "app", app, dirs_exist_ok=True,
ignore=shutil.ignore_patterns("__pycache__", "*.pyc"))
return cel
@pytest.fixture(scope="session")
def drzewo(tmp_path_factory) -> pathlib.Path:
return zloz(tmp_path_factory.mktemp("astroklient"))
@@ -0,0 +1,88 @@
"""Po czym astroklienta NIE MA — ani w odpowiedzi, ani w kodzie w obrazie.
Ta usługa dzieli kod z pełną aplikacją, więc granica między produktami nie jest
oczywista z samego patrzenia: przebiega przez listę importów w `main.py` i przez
`usun.txt`. Ten test sprawdza jej SKUTEK na złożonym drzewie — bo tylko skutek
trafia do obrazu, który komuś się oddaje.
„Bez śladu" znaczy tu również bez zaprzeczeń. Zdanie „nie ma tu generowania
tekstu" mówi wprost, że coś takiego istnieje, więc jest takim samym śladem jak
przycisk. Pierwsza wersja tego testu wyłapała trzy moje własne docstringi
wyliczające nieobecne funkcje.
"""
from __future__ import annotations
import pathlib
import re
import pytest
# Funkcje, których ten produkt nie ma — ani ich słownictwo.
GENEROWANIE = ["prompt", "LLM", "model językow", "generowanie tekstu", "sztuczn",
"anthropic", "openai", "gpt", "claude", "horoscope",
"x-ai.css", "models.js", "progress.js", "natal.js", "predictions.js"]
NIEOBECNE_EKRANY = ["Skompiluj", "Ustawienia", "Konta i uprawnien", "Eksport plików"]
MECHANIZMY = ["administrator", "kwarantann", "walidacj", "reguły przyjmowania"]
SLOWA = GENEROWANIE + NIEOBECNE_EKRANY + MECHANIZMY
WZORCE = [(r"(?<![\w-])AI(?![\w-])", "skrót nazwy funkcji")]
def _pliki(drzewo: pathlib.Path):
for f in sorted((drzewo / "app").rglob("*")):
if f.is_file() and f.suffix != ".pyc" and "__pycache__" not in str(f):
yield f
def test_the_shipped_tree_carries_no_trace(drzewo):
grzechy = []
for f in _pliki(drzewo):
try:
tresc = f.read_text(encoding="utf-8")
except UnicodeDecodeError:
continue
gdzie = f.relative_to(drzewo)
for slowo in SLOWA:
if re.search(rf"(?i)(?<![\w-]){re.escape(slowo)}", tresc):
grzechy.append(f"{gdzie}: „{slowo}")
for wzor, powod in WZORCE:
m = re.search(wzor, tresc)
if m:
grzechy.append(f"{gdzie}: „{m.group(0)}” — {powod}")
assert not grzechy, (
"\nW drzewie astroklienta padły słowa o funkcjach, których nie ma. "
"`grep` po kontenerze ma nic nie znaleźć:\n " + "\n ".join(grzechy))
def test_the_check_would_notice_if_something_slipped_in(drzewo, tmp_path):
"""Kontrola negatywna: zapora ma coś łapać.
Bez niej test wyżej przechodziłby także wtedy, gdyby lista słów była pusta
albo gdyby przeszukiwanie trafiało w niewłaściwy katalog."""
podrzucony = drzewo / "app" / "podrzucony.py"
podrzucony.write_text("# generowanie tekstu przez model\n", encoding="utf-8")
try:
with pytest.raises(AssertionError):
test_the_shipped_tree_carries_no_trace(drzewo)
finally:
podrzucony.unlink()
def test_the_removal_list_matches_what_is_actually_composed(drzewo):
"""usun.txt i main.py muszą mówić to samo.
Rozjazd między nimi znaczy albo martwy kod w obrazie (moduł usunięty
z importów, ale nie z listy), albo błąd dopiero przy uruchomieniu (plik
usunięty, a wciąż importowany). Jedno i drugie wychodzi za późno.
"""
main = (drzewo / "app" / "main.py").read_text(encoding="utf-8")
zaimportowane = set(re.findall(r"^\s{4}(\w+),$", main, re.M))
obecne = {f.stem for f in (drzewo / "app" / "ekrany").glob("*.py")
if f.stem != "__init__"}
assert zaimportowane == obecne, (
f"ekrany w drzewie: {sorted(obecne)}\n"
f"ekrany importowane: {sorted(zaimportowane)}")
def test_no_optional_module_survived(drzewo):
"""Katalogu modułów opcjonalnych nie ma wcale — most odkrywa wtedy zero."""
assert not (drzewo / "app" / "moduly").exists()
+101
View File
@@ -0,0 +1,101 @@
"""Astroklient jest produktem, a nie pełną aplikacją z wyłączonymi kawałkami.
Sprawdzamy trzy rzeczy, i wszystkie na ZŁOŻONYM drzewie, nie na kodzie w repo:
czego w nim nie ma, że to, co zostało, działa, i że nie zostało po niczym słowo.
"""
from __future__ import annotations
import json
import re
import subprocess
import sys
import pytest
SONDA = '''
import json, os, hashlib, secrets, re
sol = secrets.token_bytes(16)
dk = hashlib.scrypt(b"tajne", salt=sol, n=2**14, r=8, p=1, dklen=32)
os.environ.update(ASTROKLIENT_USERS="klient:scrypt$%s$%s" % (sol.hex(), dk.hex()),
SESSION_SECRET="t"*64, COOKIES_INSECURE="1", RATE_LIMIT_PER_MIN="0")
from starlette.testclient import TestClient
from app.main import app
from app import security, session, features
from app.clients import logic_client
c = TestClient(app, follow_redirects=False)
H = {"Cookie": f"{session.COOKIE}={security.issue_session('klient')}"}
strona = c.get("/", headers=H).text
zebrane = []
logic_client.link_crypto.call_json = lambda *a, **kw: (zebrane.append(kw.get("payload")), {})[1]
c.post("/interpret", headers=H, data={"date":"1984-04-30","time":"11:20",
"tz_offset":"2","lat":"50","lon":"19","tenant":"ktos-inny"})
print("WYNIK:" + json.dumps({
"ekrany": [f.label for f in features.SCREENS],
"uprawnienia": sorted(security._uprawnienia()),
"dziala": {a: c.get(a, headers=H).status_code
for a in ("/", "/interpret", "/timeline", "/synastry",
"/significators", "/pliki")},
"nie_ma": {a: c.get(a, headers=H).status_code
for a in ("/compile", "/settings", "/accounts", "/files",
"/horoscope/stream", "/compile/pdf")},
"nawigacja": re.findall(r"<nav>(.*?)</nav>", strona, re.S)[0].count('<a href="'),
"pula": (zebrane[0] or {}).get("tenant"),
"logowanie_dziala": bool(security.verify("klient", "tajne")),
"zle_haslo_odmowa": security.verify("klient", "inne") is None,
}))
'''
@pytest.fixture(scope="module")
def wynik(drzewo):
"""Aplikacja uruchomiona w OSOBNYM PROCESIE.
Importy Pythona są zapamiętywane w procesie, więc sprawdzanie „produktu bez
modułu" tam, gdzie moduł już się zaimportował, dałoby wynik fałszywie
pozytywny."""
r = subprocess.run([sys.executable, "-c", SONDA], cwd=drzewo, capture_output=True,
text=True, env={"PYTHONPATH": str(drzewo), "PATH": "/usr/bin:/bin"})
linia = next((l for l in r.stdout.splitlines() if l.startswith("WYNIK:")), None)
assert linia, f"sonda nie wystartowała:\n{r.stdout[-2000:]}\n{r.stderr[-2000:]}"
return json.loads(linia[len("WYNIK:"):])
def test_the_build_starts_at_all(wynik):
assert wynik["dziala"]["/"] == 200
def test_every_astrological_screen_works(wynik):
assert all(k == 200 for k in wynik["dziala"].values()), wynik["dziala"]
def test_what_it_lacks_does_not_exist_rather_than_being_refused(wynik):
"""404, nie 403 i nie 500. Odmowa z powodem jest informacją, że coś tam jest."""
assert all(k == 404 for k in wynik["nie_ma"].values()), wynik["nie_ma"]
def test_the_catalogue_describes_this_product(wynik):
assert set(wynik["ekrany"]) == {"Horoskop", "Interpretacje", "Kalendarz",
"Synastria", "Sygnifikatory", "Pliki"}
def test_no_account_can_hold_a_permission_this_product_does_not_have(wynik):
"""Uprawnienia liczą się z KATALOGU, więc opisują ten obraz. Gdyby była tu
lista wpisana ręcznie, zostałaby po usuniętych funkcjach."""
assert "ai" not in wynik["uprawnienia"]
assert "export" not in wynik["uprawnienia"]
assert "admin" not in wynik["uprawnienia"]
assert "chart" in wynik["uprawnienia"] and "significators" in wynik["uprawnienia"]
def test_navigation_has_no_dead_links(wynik):
assert wynik["nawigacja"] == len(wynik["ekrany"])
def test_login_still_tells_right_password_from_wrong(wynik):
assert wynik["logowanie_dziala"] and wynik["zle_haslo_odmowa"]
def test_the_pool_comes_from_the_session_not_from_the_form(wynik):
"""Pula jedzie z zalogowanego konta. Gdyby brała się z żądania, wystarczyłoby
podstawić cudzy login, żeby czytać cudze pliki — sonda właśnie to próbuje."""
assert wynik["pula"] == "klient"
+16
View File
@@ -0,0 +1,16 @@
moduly
ekrany/skompiluj.py
ekrany/ustawienia.py
ekrany/konta.py
ekrany/pliki.py
accounts.py
files_state.py
templates/compile.html
templates/settings.html
templates/accounts.html
templates/files.html
static/s-compile.css
static/s-settings.css
static/s-accounts.css
static/s-files.css
static/compile.js
+4
View File
@@ -599,6 +599,10 @@ class TimelineRequest(BaseModel):
to_date: str
techniques: list[str] | None = None # profection | solar_return | solar_arc
interpret: bool = False # dopnij interpretacje z bazy (1B->2B)
# Pula konta (PRE-29) — puste = cały udział. Warstwa logiczna niczego tu nie
# rozstrzyga, tylko przenosi: kto do jakiej puli ma prawo, wie warstwa
# prezentacji, bo to ona zna konta.
tenant: str = ""
@app.post("/chart/timeline")
+4
View File
@@ -8,6 +8,10 @@ from pydantic import BaseModel, Field
class QueryRequest(BaseModel):
"""To, co przychodzi z formularza (przez warstwę prezentacji)."""
# Pula konta (PRE-29) — puste = cały udział. Warstwa logiczna niczego tu nie
# rozstrzyga, tylko przenosi: kto do jakiej puli ma prawo, wie warstwa
# prezentacji, bo to ona zna konta.
tenant: str = ""
query: str = Field(..., min_length=1, description="Szukana fraza.")
field: str = Field("name", description="Po którym polu szukać.")
+6 -2
View File
@@ -14,6 +14,7 @@ from app.models import QueryRequest, QueryResponse
class QueryService:
def __init__(self, data_client: DataClient | None = None) -> None:
self._wstrzykniety = data_client is not None
self.data = data_client or DataClient()
def handle(self, req: QueryRequest) -> QueryResponse:
@@ -21,8 +22,11 @@ class QueryService:
value = req.query.strip()
key = req.field.strip().lower()
# 2) zapytanie w dół do warstwy bazodanowej
raw = self.data.search(key=key, value=value, exact=req.exact, limit=req.limit)
# 2) zapytanie w dół do warstwy bazodanowej — do PULI KONTA, jeśli podano.
# Klient budowany na żądanie, bo pula wynika z tego, kto pyta, a nie ze
# stanu procesu. Wstrzyknięty klient (testy) ma pierwszeństwo.
dane = self.data if (self._wstrzykniety or not req.tenant) else DataClient(tenant=req.tenant)
raw = dane.search(key=key, value=value, exact=req.exact, limit=req.limit)
# 3) opracowanie wyników w górę (tu można liczyć/wzbogacać/sortować)
results = raw.get("rows", [])
+3 -2
View File
@@ -4,8 +4,9 @@ from __future__ import annotations
from fastapi import Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from app import accounts as accounts_store
from app.features import Feature
from app.podstawa import (app, templates, security, features, accounts_store)
from app.podstawa import (app, templates, security, features)
# ---------------- Konta i uprawnienia (PRE-27) ----------------
# Ochrona tych tras siedzi w features.ROUTES, nie w dekoratorze — jedna mapa
@@ -94,7 +95,7 @@ def accounts_delete(request: Request, login: str = Form(...)):
# Ekran wyłącznie administracyjny — nie ma pozycji w katalogu funkcji, bo ADMIN
# nie jest uprawnieniem do nadania (patrz features.GRANTABLE).
features.zarejestruj(
nawigacja_admina=("accounts", "/accounts", "Konta"),
nawigacja=("accounts", "/accounts", "Konta", features.ADMIN),
trasy={("GET", "/accounts"): features.ADMIN,
("POST", "/accounts/create"): features.ADMIN,
("POST", "/accounts/update"): features.ADMIN,
+7 -1
View File
@@ -6,8 +6,14 @@ from fastapi.responses import HTMLResponse, RedirectResponse
import base64
import httpx
from app import files_state
from app.features import Feature
from app.podstawa import (app, templates, logic, features, files_state, _perms, _logic_error)
from app.podstawa import (app, templates, logic, features, _perms, _logic_error)
# Etykiety stanów pliku dla szablonu. Rejestrowane TUTAJ, a nie w podstawie:
# należą do tego ekranu, a produkt, który go nie ma, nie ma po co ich znać.
templates.env.globals["STATUS_LABELS"] = files_state.LABELS
# ---------------- Pliki baz (DAN-27) ----------------
# Trzy poziomy dostępu, opisane w features.ROUTES:
+9 -7
View File
@@ -45,10 +45,12 @@ class Feature:
# `kolejnosc` jest jawna, bo o zawartości decyduje lista importów w `main.py`,
# a kolejność importów nie może rządzić kolejnością zakładek w nawigacji.
SCREENS: list[Feature] = []
# Pozycje nawigacji dla administratora. Osobno od SCREENS, bo ADMIN nie jest
# uprawnieniem do nadania — a wpisane na sztywno w szablonie zostawiały nazwę
# ekranu (i martwy odsyłacz) w produkcie, który tego ekranu nie ma.
NAWIGACJA_ADMINA: list[tuple[str, str, str]] = [] # (klucz, adres, etykieta)
# Pozycje nawigacji dla uprawnień NIENADAWALNYCH — tych, których nie ma w
# GRANTABLE, więc nie da się ich przyznać z żadnego ekranu. Osobno od SCREENS
# i wraz z wymaganym uprawnieniem, żeby szablon nie musiał go wypowiadać: wpisane
# w szablonie na sztywno zostawiało nazwę ekranu i martwy odsyłacz w produkcie,
# który tego ekranu nie ma.
NAWIGACJA_SPECJALNA: list[tuple[str, str, str, str]] = [] # (klucz, adres, etykieta, uprawnienie)
EXTRAS: list[Feature] = []
ALL: tuple[Feature, ...] = ()
BY_KEY: dict[str, Feature] = {}
@@ -70,7 +72,7 @@ def _przelicz() -> None:
def zarejestruj(*, ekran: Feature | None = None, rozszerzenie: Feature | None = None,
trasy: dict[tuple[str, str], str | None] | None = None,
zasoby: dict | None = None,
nawigacja_admina: tuple[str, str, str] | None = None) -> None:
nawigacja: tuple[str, str, str, str] | None = None) -> None:
"""Zgłoszenie funkcji, jej tras i jej zasobów.
Idempotentne po kluczu: `export` zgłaszają dwa ekrany (raport do PDF-a
@@ -78,8 +80,8 @@ def zarejestruj(*, ekran: Feature | None = None, rozszerzenie: Feature | None =
for pozycja, zbior in ((ekran, SCREENS), (rozszerzenie, EXTRAS)):
if pozycja and pozycja.key not in {f.key for f in zbior}:
zbior.append(pozycja)
if nawigacja_admina and nawigacja_admina[0] not in {p[0] for p in NAWIGACJA_ADMINA}:
NAWIGACJA_ADMINA.append(nawigacja_admina)
if nawigacja and nawigacja[0] not in {p[0] for p in NAWIGACJA_SPECJALNA}:
NAWIGACJA_SPECJALNA.append(nawigacja)
ROUTES.update(trasy or {})
STATIC.update(zasoby or {})
_przelicz()
+9 -4
View File
@@ -24,12 +24,10 @@ from fastapi.responses import Response
from fastapi.templating import Jinja2Templates
from app import accounts as accounts_store
from app import rozszerzenia
from app import chartwheel as chartwheel_mod
from app import features
from app import session
from app import files_state
from app.house_systems import HOUSE_SYSTEMS, LIMITED as HOUSE_LIMITED, label as house_label
from app import geocode, security
@@ -145,13 +143,16 @@ def static_file(nazwa: str):
templates.env.globals["static"] = static_url
templates.env.globals["nav_admin"] = lambda: features.NAWIGACJA_ADMINA
# Pozycje specjalne przefiltrowane uprawnieniem — szablon dostaje gotową listę
# i nie musi wiedzieć, czego która wymaga.
templates.env.globals["nav_extra"] = lambda request: [
(k, a, e) for k, a, e, u in features.NAWIGACJA_SPECJALNA
if features.can(_perms(request), u)]
# Katalog systemów domów — jedno źródło nazw dla wszystkich szablonów.
templates.env.globals["HOUSE_SYSTEMS"] = HOUSE_SYSTEMS
templates.env.globals["house_label"] = house_label
templates.env.globals["HOUSE_LIMITED"] = HOUSE_LIMITED
templates.env.globals["WHEEL_ORIENTATIONS"] = chartwheel_mod.ORIENTATIONS
templates.env.globals["STATUS_LABELS"] = files_state.LABELS
def _perms(request: Request) -> frozenset[str]:
@@ -169,6 +170,10 @@ templates.env.globals["nav_screens"] = lambda request: features.screens_for(_per
templates.env.globals["current_user"] = lambda request: getattr(request.state, "user", "") or ""
templates.env.globals["auth_on"] = security.auth_enabled
templates.env.globals["can"] = lambda request, key: key in _perms(request)
# Czy to konto ma uprawnienie NIENADAWALNE — czyli takie, którego nie da się
# przyznać z żadnego ekranu. Nazwa neutralna, bo szablon jest wspólny.
templates.env.globals["widok_diagnostyczny"] = lambda request: features.can(
_perms(request), features.ADMIN)
def _limit_options(request: Request, opts: dict) -> dict:
+1 -1
View File
@@ -175,7 +175,7 @@ def refresh(dane: dict, *, now: float | None = None) -> str | None:
def cookie_params(secure: bool = True) -> dict:
"""Ustawienia ciasteczka wspólne dla ustawiania i kasowania.
"""Parametry ciasteczka wspólne dla ustawiania i kasowania.
httponly JavaScript nie ma po co go czytać, a bez tego jeden wstrzyknięty
skrypt wynosi sesję. samesite=strict odcina większość ataków polegających
@@ -1,4 +1,4 @@
{# Opcje obliczeń — WSPÓLNE dla „Horoskop" (/) i „Skompiluj" (/compile). Jeden plik,
{# Opcje obliczeń — WSPÓLNE dla „Horoskop" (/) i raportu zbiorczego (/compile). Jeden plik,
bo dodawaliśmy je już 3× w dwóch formularzach i groziło rozjechaniem (regresja
podsumowania). Wszystkie synchronizowane między zakładkami przez formsync.js. #}
{% if can(request, 'advanced_calc') %}
@@ -1,4 +1,4 @@
{# Tabele wyniku horoskopu — używane przez „Skompiluj" (/compile), żeby
{# Tabele wyniku horoskopu — używane przez raportu zbiorczego (/compile), żeby
podsumowanie pokazywało dokładnie to, co policzono (regresja 2026-07-28:
podsumowanie miało braki). Skopiowane 1:1 z sekcji tabel na „Horoskop" (/).
TODO: przełączyć też chart.html na ten include, żeby oba widoki nie mogły się
+8 -10
View File
@@ -28,14 +28,12 @@
{% for tab in nav_screens(request) %}
<a href="{{ tab.href }}" class="{{ 'active' if nav_active is defined and nav_active == tab.key else '' }}">{{ tab.label }}</a>
{% endfor %}
{# Pozycje administracyjne z REJESTRU, nie z listy w szablonie: wpisany tu
na sztywno odsyłacz zostawał w produkcie, który tego ekranu nie ma —
{# Pozycje specjalne z REJESTRU, już przefiltrowane uprawnieniem. Wpisany
tu na sztywno odsyłacz zostawał w produkcie, który tego ekranu nie ma —
razem z nazwą ekranu i martwym adresem. #}
{% if can(request, 'admin') %}
{% for klucz, adres, etykieta in nav_admin() %}
{% for klucz, adres, etykieta in nav_extra(request) %}
<a href="{{ adres }}" class="{{ 'active' if nav_active is defined and nav_active == klucz else '' }}">{{ etykieta }}</a>
{% endfor %}
{% endif %}
{# Wyjście na końcu paska i odsunięte: to nie jest zakładka, tylko
czynność — pomylenie go z zakładką kosztuje przypadkowe wylogowania.
Pokazujemy też KTO jest zalogowany, bo bez tego przycisk wyjścia jest
@@ -52,11 +50,11 @@
</nav>
</header>
{% block content %}{% endblock %}
{# Stopka wymieniała warstwy architektury i nazywała całość „widokiem
testowym" — czyli mówiła, że to nie jest gotowy produkt i że za nim stoi
łańcuch usług. Dla konta ograniczonego program ma wyglądać na kompletny,
więc ten opis widzi wyłącznie administrator. #}
{% if can(request, 'admin') %}
{# Stopka wymienia warstwy architektury i nazywa całość „widokiem testowym" —
czyli mówi, że to nie jest gotowy produkt i że za nim stoi łańcuch usług.
Widzi ją wyłącznie konto z uprawnieniem nienadawalnym; dla pozostałych
program ma wyglądać na kompletny. #}
{% if widok_diagnostyczny(request) %}
<footer class="foot">
prezentacja → logika → dane · widok testowy
</footer>