Compare commits

..

2 Commits

Author SHA1 Message Date
gitea d4a37aca52 feat(presentation): wyszukiwarka lokalizacji + mapa (OSM/Leaflet)
QoL 2/2: pole „Szukaj miejsca" (nazwa/adres/POI) + interaktywna mapa na
formularzach horoskopu / interpretacji / kalendarza. Bez klucza API.

- geocode.py: proxy OSM/Nominatim PO STRONIE SERWERA (poprawny User-Agent,
  throttling ~1 req/s, cache TTL 1h) — endpointy /geocode i /reverse.
  Wolanie z serwera, nie z przegladarki: latwiej trzymac polityke Nominatim
  i dziala niezaleznie od secure-context (http://<ip>).
- _location_picker.html: wspolny partial (search + wyniki + mapa + atrybucja),
  wpiety includem do 3 formularzy z polami lat/lon.
- geo.js: Leaflet — wyszukiwanie (debounce), klik na wynik ustawia lat/lon i
  centruje mape, klik/drag pineski ustawia wspolrzedne + /reverse pokazuje
  nazwe; sync z „Tu i teraz" (now.js emituje event astrololo:coords).
- Leaflet 1.9.4 vendorowany lokalnie (static/vendor/leaflet, BSD-2-Clause,
  permisywny) — niezaleznosc od CDN; kafelki mapy z OSM. Marker jako divIcon
  (bez plikow PNG).
- styles.css: style pod ciemny motyw.

Zweryfikowane w przegladarce: domyslny widok (pineska na Szpitalu Barlickiego),
wyszukanie „Wawel Krakow" -> lista -> klik ustawia 50.0547/19.9361 i przesuwa
mape, klik w mape ustawia wspolrzedne + reverse wypelnia nazwe. Zero bledow
w konsoli. /geocode i /reverse zwracaja szpital; cache dziala.

Uwaga wdrozeniowa: pod prezentacji potrzebuje egressu do
nominatim.openstreetmap.org; przegladarki — do tile.openstreetmap.org.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-21 03:26:03 +02:00
gitea e39c31618d feat(presentation): domyslna lokalizacja = Szpital Barlickiego, Lodz
QoL: formularze (horoskop / interpretacje / kalendarz) maja wstepnie
wpisana lokalizacje urodzenia wlasciciela — Szpital Barlickiego w Lodzi
(51.7739N, 19.4829E; potwierdzone reverse-geokodowaniem OSM: Kopcinskiego
22/28). Nie trzeba jej wpisywac za kazdym razem.

- config.py: jedno zrodlo prawdy (DEFAULT_LAT/LON/LABEL, nadpisywalne ENV)
  + helper default_form().
- main.py: GET wstrzykuje default_form() + location_label do 3 formularzy
  z polami lokalizacji (significators pominiete — nie ma tam lat/lon).
- szablony: dyskretna podpowiedz z nazwa lokalizacji, widoczna tylko na
  czystym formularzu (po POST znika, wygrywa wpisana wartosc).
- „Tu i teraz" nadal nadpisuje domyslne wspolrzedne geolokalizacja.

Zweryfikowane TestClientem: 3 strony renderuja 51.7739/19.4829 + etykiete;
POST z innymi wspolrzednymi je zachowuje i chowa podpowiedz.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-21 03:12:59 +02:00
54 changed files with 122 additions and 5741 deletions
-32
View File
@@ -1,32 +0,0 @@
name: build-swisseph
# Osobny pipeline dla silnika B (Swiss Ephemeris, AGPL) — celowo ODDZIELONY od
# głównego build.yaml (data/logic/presentation). Buduje się tylko, gdy zmienia się
# sam silnik, i nie miesza obrazu AGPL do pipeline'u permisywnego produktu.
#
# Obraz konsumuje profil deployu `astrololo-swisseph` w repo `deploy`.
on:
push:
branches: [master]
paths:
- 'services/engine-swisseph/**'
- '.gitea/workflows/build-swisseph.yaml'
workflow_dispatch: {} # ręczne odpalenie (bootstrap pierwszego obrazu)
jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Login
run: echo "${{ secrets.REGISTRY_TOKEN }}" | docker login gitea.czernobog.pl -u gitea --password-stdin
- name: Build & push engine-swisseph (AGPL, izolowany)
run: |
TAG=${GITHUB_SHA::8}
IMG=gitea.czernobog.pl/gitea/astrololo-engine-swisseph
# Obraz kompiluje pyswisseph ze źródeł (brak wheeli dla cp312) — build jest
# zarazem realnym testem Dockerfile'a.
docker build -t $IMG:$TAG -t $IMG:latest ./services/engine-swisseph
docker push $IMG:$TAG
docker push $IMG:latest
echo "Zbudowano i wypchnięto: $IMG:$TAG (+ latest)"
-142
View File
@@ -1,142 +0,0 @@
name: Testy
# Odpala się przy każdym pushu (dowolna gałąź) oraz dla pull requestów do master.
on:
push:
pull_request:
branches: [master]
jobs:
logic-tests:
name: Testy warstwy logicznej (silnik)
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Python 3.12 (jak w obrazach Dockera)
uses: actions/setup-python@v5
with:
python-version: "3.12"
cache: pip
cache-dependency-path: services/logic/requirements-dev.txt
# Jądro efemeryd JPL (de421.bsp, ~17 MB). Cache'ujemy je między runami.
- name: Cache jądra efemeryd
uses: actions/cache@v4
with:
path: services/logic/.ephemeris
key: ephemeris-de421
# Pobieramy jawnie (a nie licząc na auto-pobranie przez Skyfield), żeby
# brak jądra był twardym błędem, a nie cichym pomijaniem testów.
- name: Pobierz jądro efemeryd (gdy brak w cache)
run: |
mkdir -p services/logic/.ephemeris
if [ ! -s services/logic/.ephemeris/de421.bsp ]; then
curl -fSL --retry 3 --max-time 300 \
-o services/logic/.ephemeris/de421.bsp \
https://ssd.jpl.nasa.gov/ftp/eph/planets/bsp/de421.bsp
fi
ls -lh services/logic/.ephemeris/de421.bsp
- name: Instalacja zależności
run: pip install -r services/logic/requirements-dev.txt
# CI=true (ustawiane przez GitHub) sprawia, że brak silnika = błąd,
# a nie pominięcie — patrz tests/conftest.py.
- name: Testy (pytest)
working-directory: services/logic
env:
PYTHONPATH: .
EPHEMERIS_DIR: ${{ github.workspace }}/services/logic/.ephemeris
run: pytest tests -q -rs
presentation-tests:
name: Testy warstwy prezentacji (dostęp do baz)
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/presentation/requirements-dev.txt
- name: Instalacja zależności
run: pip install -r services/presentation/requirements-dev.txt
# Bramka chroniąca oryginalne bazy — nietestowany kod ochronny jest gorszy
# niż jego brak, bo daje złudzenie zabezpieczenia.
- name: Testy (pytest)
working-directory: services/presentation
env:
PYTHONPATH: .
run: pytest tests -q -rs
swisseph-image:
name: Build obrazu silnika B (swisseph)
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
# Obraz kompiluje pyswisseph ze źródeł (brak wheeli dla cp312), więc ten
# build jest realnym testem Dockerfile'a — nie tylko pobraniem paczek.
- name: docker build
run: docker build -t astrololo/engine-swisseph:ci services/engine-swisseph
# Test biegnie WEWNĄTRZ obrazu, bez sieci i bez kontenera w tle. Poprzednia
# wersja startowała kontener w tle (--name swe) i pukała curl-em w
# localhost:8003 — co miało dwie wady:
# 1. nie sprzątała kontenera, więc każdy kolejny przebieg padał na
# konflikcie nazwy (Conflict. The container name "/swe" is already in use),
# 2. job Gitea Actions sam działa w kontenerze, a -p publikuje port na
# HOŚCIE — więc localhost joba to nie ten sam localhost.
# Wywołanie funkcji endpointów wprost omija oba problemy, a sprawdza to samo:
# obraz się zbudował, pyswisseph liczy, kontrakt /positions się zgadza.
# --rm gwarantuje, że nic nie zostaje po przebiegu.
- name: Smoke test (health + pozycje) wewnątrz obrazu
run: |
docker run --rm astrololo/engine-swisseph:ci python - <<'PY'
from datetime import datetime, timezone
from app.main import DEFAULT_OBJECTS, PositionsRequest, health, positions
h = health()
assert h["status"] == "ok", h
print("health:", h)
# Horoskop referencyjny (30.04.1984) — ten sam, na którym opieramy testy
# silnika własnego; sprawdzamy, że silnik B faktycznie liczy.
req = PositionsRequest(when_utc=datetime(1984, 4, 30, 9, 20, tzinfo=timezone.utc),
lat=50.0647, lon=19.9450)
out = positions(req)
by = {p["name"]: p for p in out["positions"]}
assert out["engine"] == "swisseph", out["engine"]
assert len(by) == len(DEFAULT_OBJECTS), sorted(by)
sun = by["Sun"]["longitude"]
assert 39.5 < sun < 41.0, f"Slonce poza oczekiwanym zakresem: {sun}"
nn, sn = by["North Node"]["longitude"], by["South Node"]["longitude"]
assert abs(((sn - nn) % 360.0) - 180.0) < 1e-6, (nn, sn)
print(f"Sun={sun:.4f} NN={nn:.4f} obiektow={len(by)}")
print("SMOKE OK")
PY
# Sprzątanie po POPRZEDNICH przebiegach starej wersji workflow, która
# zostawiała kontener „swe" na runnerze i blokowała nazwę. Nowa wersja
# kontenera w tle nie tworzy, więc to tylko jednorazowe uprzątnięcie.
- name: Usuń osierocony kontener ze starych przebiegów
if: always()
run: docker rm -f swe 2>/dev/null || true
compile-all:
name: Kontrola składni wszystkich warstw
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
# Sam kompilator — bez instalowania zależności warstw (w tym AGPL-owego
# silnika swisseph, który nie wchodzi do produktu).
- name: py_compile
run: python -m compileall -q services
+65
View File
@@ -0,0 +1,65 @@
name: Testy
# Odpala się przy każdym pushu (dowolna gałąź) oraz dla pull requestów do master.
on:
push:
pull_request:
branches: [master]
jobs:
logic-tests:
name: Testy warstwy logicznej (silnik)
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Python 3.12 (jak w obrazach Dockera)
uses: actions/setup-python@v5
with:
python-version: "3.12"
cache: pip
cache-dependency-path: services/logic/requirements-dev.txt
# Jądro efemeryd JPL (de421.bsp, ~17 MB). Cache'ujemy je między runami.
- name: Cache jądra efemeryd
uses: actions/cache@v4
with:
path: services/logic/.ephemeris
key: ephemeris-de421
# Pobieramy jawnie (a nie licząc na auto-pobranie przez Skyfield), żeby
# brak jądra był twardym błędem, a nie cichym pomijaniem testów.
- name: Pobierz jądro efemeryd (gdy brak w cache)
run: |
mkdir -p services/logic/.ephemeris
if [ ! -s services/logic/.ephemeris/de421.bsp ]; then
curl -fSL --retry 3 --max-time 300 \
-o services/logic/.ephemeris/de421.bsp \
https://ssd.jpl.nasa.gov/ftp/eph/planets/bsp/de421.bsp
fi
ls -lh services/logic/.ephemeris/de421.bsp
- name: Instalacja zależności
run: pip install -r services/logic/requirements-dev.txt
# CI=true (ustawiane przez GitHub) sprawia, że brak silnika = błąd,
# a nie pominięcie — patrz tests/conftest.py.
- name: Testy (pytest)
working-directory: services/logic
env:
PYTHONPATH: .
EPHEMERIS_DIR: ${{ github.workspace }}/services/logic/.ephemeris
run: pytest tests -q -rs
compile-all:
name: Kontrola składni wszystkich warstw
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
# Sam kompilator — bez instalowania zależności warstw (w tym AGPL-owego
# silnika swisseph, który nie wchodzi do produktu).
- name: py_compile
run: python -m compileall -q services
Binary file not shown.
-296
View File
@@ -1,296 +0,0 @@
# Wdrożenie PRE-16 — HTTPS na wejściu i szyfrowanie łączy między warstwami
Instrukcja krok po kroku. **Kolejność ma znaczenie** — punkt „Dlaczego taka
kolejność" niżej tłumaczy, co się stanie, jeśli ją zamienić.
Dotyczy dwóch pull requestów:
| Repo | PR | Co wnosi |
|---|---|---|
| `gitea/astrololo` | [#21](https://gitea.czernobog.pl/gitea/astrololo/pulls/21) | kod: szyfrowanie łączy, limit żądań za proxy |
| `gitea/deploy` | [#4](https://gitea.czernobog.pl/gitea/deploy/pulls/4) | manifesty: Ingress, certyfikat, klucze łączy |
---
## Co się właściwie zmienia
**Na wejściu do aplikacji.** Dotąd logowanie szło przez HTTP Basic po zwykłym
http — czyli hasło leciało siecią w postaci trywialnej do podsłuchania (base64 to
nie szyfrowanie). Po zmianie wejście jest po https, a http odsyła na https.
Przy okazji **odblokowują się dwie funkcje zepsute dziś z tego samego powodu**:
geolokalizacja („Tu i teraz") i kopiowanie promptu do schowka działają wyłącznie
w tzw. secure context i po http po prostu odmawiały.
**Między warstwami.** Prezentacja, logika i dane rozmawiały ze sobą otwartym
tekstem wewnątrz klastra. Token międzywarstwowy mówił *kto* pyta, ale nie ukrywał
*czego dotyczy odpowiedź* — a płyną nią surowe wiersze oryginalnych baz. Teraz
każde ciało żądania i odpowiedzi jest szyfrowane **AES-256-GCM**, osobnym kluczem
na każdą parę rozmówców.
**Wejście na świat pozostaje jedno: prompt do modelu.** Ta zmiana niczego tu nie
rusza — dotyczy wyłącznie ruchu wewnątrz sieci i wejścia z przeglądarki.
---
## Zanim zaczniesz — stan wyjściowy
```bash
kubectl -n astrololo get deploy,svc
kubectl -n astrololo get secret # powinny być: astrololo-auth, gitea-registry
kubectl -n kube-system get svc traefik -o jsonpath='{.status.loadBalancer.ingress[*].ip}'; echo
```
Zanotuj adres Traefika — będzie potrzebny w kroku 3. Sprawdź też, czy działa
aplikacja w obecnej postaci (przez NodePort), żeby mieć punkt odniesienia.
---
## Krok 1 — sekret z kluczami łączy
**Przed czymkolwiek innym.** Klucze muszą istnieć, zanim pody spróbują wstać
z nową konfiguracją, bo bez nich celowo **nie wystartują**.
```bash
kubectl -n astrololo create secret generic astrololo-link \
--from-literal=LINK_KEY_PRESENTATION_LOGIC="$(openssl rand -hex 32)" \
--from-literal=LINK_KEY_LOGIC_DATA="$(openssl rand -hex 32)"
```
Kluczy nikt nigdy nie musi oglądać — służą tylko usługom. Nie ma ich w repo
GitOps i **nie ma ich tam wkładać**: cokolwiek trafi do gita, zostaje w historii
na zawsze.
Dwa osobne klucze to nie ozdobnik. Przejęcie klucza prezentacji nie daje dostępu
do warstwy danych, gdzie leżą całe bazy. Logika dostaje oba, bo rozmawia w obie
strony; prezentacja i dane dostają wyłącznie swój.
Sprawdź:
```bash
kubectl -n astrololo get secret astrololo-link -o jsonpath='{.data}' | tr ',' '\n'
# oczekiwane: dwa klucze, każdy 64 znaki po odkodowaniu (32 bajty)
```
---
## Krok 2 — cert-manager
Jednorazowo, na cały klaster:
```bash
kubectl apply -f https://github.com/cert-manager/cert-manager/releases/download/v1.21.0/cert-manager.yaml
kubectl -n cert-manager rollout status deploy/cert-manager deploy/cert-manager-webhook --timeout=180s
```
Poczekaj, aż **webhook** będzie gotowy — dopóki nie wstanie, tworzenie obiektów
`Certificate` kończy się błędem połączenia i wygląda jak zepsuty manifest.
Sprawdź:
```bash
kubectl get crd | grep cert-manager | head -3 # muszą się pojawić
```
> **Dlaczego własne CA, a nie Let's Encrypt.** Klaster stoi w LAN (Traefik trzyma
> LoadBalancera na adresach 192.168.1.x), więc walidacja HTTP-01 nie ma jak dojść
> z internetu, a DNS-01 wymagałby trzymania w klastrze tokena API do domeny.
> Własne CA nie potrzebuje niczego z zewnątrz i odnawia certyfikaty samo. Cena:
> raz na urządzenie importujesz korzeń (krok 6).
---
## Krok 3 — DNS
Wpis `astrololo.czernobog.pl` → adres Traefika z kroku „stan wyjściowy”.
W routerze, lokalnym DNS-ie albo doraźnie w `/etc/hosts`:
```bash
echo "192.168.1.73 astrololo.czernobog.pl" | sudo tee -a /etc/hosts
```
**To nie jest krok opcjonalny.** Service `presentation` przestaje być NodePortem
(był drugą, nieszyfrowaną drogą do aplikacji — czyli obejściem całego PRE-16),
więc po wdrożeniu manifestów nazwa jest jedynym wejściem. Awaryjnie zawsze zostaje:
```bash
kubectl -n astrololo port-forward svc/presentation 8000:8000 # http://localhost:8000
```
---
## Krok 4 — merge PR-a aplikacji (astrololo #21)
Teraz, **przed** manifestami.
```bash
tea pr merge --login gitea --repo gitea/astrololo 21
```
Po merge'u CI zbuduje obrazy, a image-updater sam podbije tagi w repo `deploy`,
skąd ArgoCD wymieni pody. Poczekaj, aż to się przetoczy:
```bash
kubectl -n astrololo rollout status deploy/presentation deploy/logic deploy/data
kubectl -n astrololo get pods -o jsonpath='{range .items[*]}{.spec.containers[0].image}{"\n"}{end}'
```
Na tym etapie **nic się jeszcze nie szyfruje** — nowy kod to potrafi, ale zmienne
z kluczami dokłada dopiero PR do `deploy`. Aplikacja działa dokładnie jak dotąd.
To celowe: chcemy, żeby *cała* obsada podów umiała szyfrować, zanim ktokolwiek
tego zażąda.
---
## Krok 5 — merge PR-a manifestów (deploy #4)
```bash
tea pr merge --login gitea --repo gitea/deploy 4
```
ArgoCD zsynchronizuje się sam (`automated`, `selfHeal`). Wjeżdża naraz: Ingress,
certyfikat, zmienne z kluczami, `TRUST_PROXY` i zdjęcie NodePortu.
```bash
kubectl -n argocd get application astrololo
kubectl -n astrololo rollout status deploy/presentation deploy/logic deploy/data
kubectl -n astrololo get certificate # astrololo-ca i astrololo-tls: READY=True
```
> **Spodziewaj się kilkudziesięciu sekund błędów w trakcie.** Pody wymieniają się
> po kolei, więc przez chwilę stara prezentacja (jeszcze bez klucza) rozmawia
> z nową logiką (już z kluczem) i dostaje odmowę. To zamierzone: alternatywą byłby
> tryb „przyjmuj i szyfrowane, i jawne”, który zwykle zostaje włączony na zawsze.
Merge nie cofnie tagów obrazów — PR dotyka w `kustomization.yaml` wyłącznie listy
`resources`, nie bloku `images`, więc git złoży to z nowszymi tagami z mastera.
---
## Krok 6 — zaufanie do własnego CA (raz na urządzenie)
Bez tego przeglądarka pokaże ostrzeżenie o certyfikacie. Korzeń jest ważny 10 lat,
więc robisz to raz:
```bash
kubectl -n astrololo get secret astrololo-ca -o jsonpath='{.data.ca\.crt}' \
| base64 -d > astrololo-ca.crt
# macOS — do systemowego zaufania (poprosi o hasło administratora)
sudo security add-trusted-cert -d -r trustRoot \
-k /Library/Keychains/System.keychain astrololo-ca.crt
# Linux (Debian/Ubuntu)
sudo cp astrololo-ca.crt /usr/local/share/ca-certificates/ && sudo update-ca-certificates
```
Firefox ma **własny** magazyn certyfikatów — import przez *Ustawienia →
Prywatność i bezpieczeństwo → Wyświetl certyfikaty → Organy certyfikacji*.
---
## Krok 7 — sprawdzenie, że działa to, co miało zadziałać
### Wejście po https
```bash
curl -sI http://astrololo.czernobog.pl/ | head -2 # 301 → https
curl -s -o /dev/null -w "bez hasła: %{http_code}\n" https://astrololo.czernobog.pl/
curl -s -o /dev/null -w "z hasłem: %{http_code}\n" -u astrololo:'<hasło>' https://astrololo.czernobog.pl/
curl -sI -u astrololo:'<hasło>' https://astrololo.czernobog.pl/ | grep -i strict-transport
```
Oczekiwane: **301**, **401**, **200**, nagłówek HSTS obecny. Brak ostrzeżenia
o certyfikacie w przeglądarce oznacza, że krok 6 się udał.
### W przeglądarce
Kliknij **„Tu i teraz"** — powinno pobrać lokalizację (po http odmawiało).
Wygeneruj prompt i kliknij **kopiuj** — schowek powinien zadziałać bez obejść.
### Szyfrowanie łączy — sprawdzenie wprost
Najmocniejszy test to próba obejścia. Z wnętrza klastra, **bez klucza**:
```bash
kubectl -n astrololo exec deploy/presentation -- \
python -c "
import httpx, os
r = httpx.post('http://logic:8001/chart/report',
json={'when_utc':'1984-04-30T09:20:00+00:00','lat':50.06,'lon':19.94},
headers={'X-Astrololo-Token': os.environ['INTERNAL_TOKEN']})
print(r.status_code, r.text[:120])
"
```
Oczekiwane: **400** i `Łącze międzywarstwowe wymaga szyfrowania.` Zwróć uwagę, że
żądanie miało **prawidłowy token** — sam token już nie wystarcza, i o to chodziło.
To samo w dół, do warstwy danych:
```bash
kubectl -n astrololo exec deploy/logic -- \
python -c "
import httpx, os
r = httpx.post('http://data:8002/search',
json={'key':'significator','value':'[Sat','exact':False,'limit':5},
headers={'X-Astrololo-Token': os.environ['INTERNAL_TOKEN']})
print(r.status_code, r.text[:120])
"
```
### Logi startowe
```bash
kubectl -n astrololo logs deploy/logic | grep -i "łącze\|UWAGA"
```
Powinno być `łącze szyfrowane (AES-256-GCM…)`. Jeśli widzisz ostrzeżenie
o rozmowie **jawnym tekstem** — klucz nie doszedł do poda.
---
## Dlaczego taka kolejność
| Kolejność | Skutek zamiany |
|---|---|
| Sekret **przed** manifestami | `LINK_ENCRYPTION_REQUIRED=true` bez klucza celowo wywraca start. Pody wpadną w CrashLoop i będą tak siedzieć do czasu utworzenia sekretu. |
| cert-manager **przed** manifestami | API odrzuci `Certificate`/`Issuer` jako nieznane rodzaje zasobów, ArgoCD pokaże aplikację jako niezsynchronizowaną i sam tego nie naprawi. |
| DNS **przed** manifestami | NodePort znika razem z nimi. Bez wpisu DNS zostaje tylko `port-forward`. |
| Aplikacja **przed** manifestami | Odwrotnie: manifesty włączyłyby szyfrowanie na obrazach, które go nie znają — wszystkie żądania kończyłyby się odmową do czasu przebudowy obrazów. |
Fail-closed w obie strony jest zamierzony. Usługa, która wstała i **po cichu nie
szyfruje**, jest gorsza niż pod w CrashLoop — awarii nie widać, a bazy jadą
otwartym tekstem.
---
## Wycofanie
Manifestów: `git revert` merge'a w `deploy` — ArgoCD samo wróci do NodePortu
i ruchu bez szyfrowania. Kod aplikacji **nie wymaga wycofania**: bez zmiennych
`LINK_KEY_*` moduł przepuszcza ruch jak dotąd (i głośno o tym mówi w logach).
Certyfikat i CA zostają w namespace; usunięcie: `kubectl -n astrololo delete
certificate astrololo-ca astrololo-tls`. cert-managera można zostawić — nie
przeszkadza.
---
## Gdy coś nie gra
| Objaw | Przyczyna | Co zrobić |
|---|---|---|
| Pody w `CrashLoopBackOff`, w logach `LINK_ENCRYPTION_REQUIRED … nie ustawiony` | brak sekretu `astrololo-link` | krok 1, potem `rollout restart` |
| `400 Łącze międzywarstwowe wymaga szyfrowania` przy normalnym korzystaniu | jedna warstwa ma klucz, druga nie (albo trwa rollout) | `rollout status`; sprawdź, czy wszystkie trzy pody mają zmienną |
| `400 Nie udało się odczytać zaszyfrowanego żądania` | klucze po obu stronach łącza są **różne** | wymień sekret i zrestartuj **wszystkie trzy** naraz |
| `Certificate` stoi w `READY=False` | webhook cert-managera jeszcze nie wstał | `kubectl -n cert-manager get pods`, poczekaj i sprawdź `kubectl -n astrololo describe certificate astrololo-tls` |
| Przeglądarka: „połączenie nie jest prywatne” | korzeń CA nieimportowany na tym urządzeniu | krok 6 (pamiętaj, że Firefox ma osobny magazyn) |
| `404` z Traefika pod adresem aplikacji | DNS wskazuje gdzie indziej niż LoadBalancer Traefika | porównaj `dig +short astrololo.czernobog.pl` z adresem z kroku „stan wyjściowy” |
| Limit żądań odcina wszystkich naraz | brak `TRUST_PROXY=true` — cały ruch liczony jako jeden klient | sprawdź zmienną w `deploy/presentation` |
---
## Czego to nie załatwia
- **Szyfrowane są ciała żądań, nie nagłówki.** Ścieżka (`/search`) i token
międzywarstwowy jadą czytelnie. Sam token nikomu nic nie daje — bez klucza łącza
każde żądanie kończy się odmową — ale metadanych to nie ukrywa. Pełne ukrycie
wymagałoby mTLS.
- **Własne CA to nie publiczne zaufanie.** Każde nowe urządzenie wymaga importu
korzenia. Gdyby aplikacja miała kiedyś wyjść na świat, właściwą drogą jest
Let's Encrypt przez DNS-01.
- **NFS z plikami baz** stoi obok aplikacji — kto ma dostęp do share'u, bierze
pliki z pominięciem wszystkich powyższych zabezpieczeń. Do zamknięcia po stronie
infrastruktury (eksport tylko dla IP węzłów, `root_squash`, najlepiej read-only).
- **Sekrety w etcd** są tylko zakodowane base64. Docelowo: szyfrowanie etcd
at-rest albo Sealed Secrets / SOPS.
-466
View File
@@ -1,466 +0,0 @@
"""Szyfrowanie łączy między warstwami (PRE-16 / LOG-33).
Do tej pory warstwy rozmawiały ze sobą zwykłym HTTP-em wewnątrz klastra. Token
międzywarstwowy (LOG-32) mówił KTO pyta, ale nie ukrywał CZEGO dotyczy odpowiedź
— a płyną nią surowe wiersze oryginalnych baz interpretacyjnych, czyli rdzeń
produktu. Kto podsłuchał ruch wewnątrz sieci (drugi pod, port mirror na switchu,
zrzut z węzła), miał je w całości.
Ten moduł zamyka tę drogę: **AES-256-GCM** na ciele każdego żądania i odpowiedzi.
GCM daje jednocześnie poufność i uwierzytelnienie — cudzy albo podmieniony bajt
nie odszyfruje się w ogóle, więc nie ma osobnego problemu „zaszyfrowane, ale
podatne na modyfikację".
**Dwa niezależne klucze**, po jednym na parę rozmówców:
* ``LINK_KEY_PRESENTATION_LOGIC`` — prezentacja ↔ logika,
* ``LINK_KEY_LOGIC_DATA`` — logika ↔ dane.
Dzięki temu przejęcie klucza prezentacji nie daje dostępu do warstwy danych,
gdzie leżą całe bazy. Logika trzyma oba, bo rozmawia w obie strony.
Z każdego klucza łącza wyprowadzamy **osobne podklucze na kierunek** (HKDF).
Żądanie i odpowiedź nigdy nie szyfrują się tym samym kluczem, więc powtórzenie
losowej jednorazówki w jedną stronę nie osłabia drugiej.
Format ramki (bo strumień odpowiedzi może iść kawałkami — patrz okno postępu):
[4 bajty długości][magia "AL1"][12 bajtów jednorazówki][szyfrogram + znacznik]
Do materiału uwierzytelnianego (AAD) wchodzą kierunek, ścieżka, znacznik czasu
i numer ramki. Skutek: ramki nie da się przekleić do innego endpointu, odtworzyć
po czasie (dopuszczalny poślizg ``MAX_SKEW``) ani przestawić w strumieniu.
Bez ustawionego klucza moduł **przepuszcza ruch otwartym tekstem** (dev, zgodność
wstecz) i krzyczy o tym przy starcie. Gdy klucz JEST ustawiony, warstwa serwerowa
działa fail-closed: nieszyfrowane żądanie dostaje odmowę, żeby przypadkowa
regresja po stronie klienta nie oznaczała cichego powrotu do jawnego ruchu.
"""
from __future__ import annotations
import base64
import binascii
import logging
import os
import struct
import time
from typing import Iterable, Iterator
from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
log = logging.getLogger("astrololo.link")
MAGIC = b"AL1"
VERSION = "v1"
NONCE_BYTES = 12
KEY_BYTES = 32 # AES-256
LENGTH_PREFIX = 4
MAX_FRAME = 64 * 1024 * 1024 # zapora przed alokacją z podanej długości
MAX_SKEW_SECONDS = 300.0
HEADER_ENC = "X-Astrololo-Enc"
HEADER_TS = "X-Astrololo-Enc-Ts"
CONTENT_TYPE = "application/vnd.astrololo.enc"
ENV_PRESENTATION_LOGIC = "LINK_KEY_PRESENTATION_LOGIC"
ENV_LOGIC_DATA = "LINK_KEY_LOGIC_DATA"
ENV_REQUIRED = "LINK_ENCRYPTION_REQUIRED"
REQUEST, RESPONSE = b"req", b"res"
# Sondy k8s pukają tu bez klucza i tak ma zostać — inaczej pierwsza literówka
# w sekrecie kładłaby pody zamiast pokazać błąd w aplikacji.
PUBLIC_PATHS = frozenset({"/health"})
class LinkError(Exception):
"""Cokolwiek poszło nie tak z kopertą — celowo bez szczegółów na zewnątrz."""
# --------------------------------------------------------------------- klucze
def parse_key(raw: str) -> bytes:
"""Klucz z konfiguracji: hex (64 znaki) albo base64. Zawsze 32 bajty."""
text = raw.strip()
if not text:
raise LinkError("pusty klucz łącza")
try:
key = bytes.fromhex(text)
except ValueError:
try:
key = base64.b64decode(text, validate=True)
except (binascii.Error, ValueError) as exc:
raise LinkError("klucz łącza nie jest ani hexem, ani base64") from exc
if len(key) != KEY_BYTES:
raise LinkError(
f"klucz łącza ma {len(key)} B zamiast {KEY_BYTES} — wygeneruj przez "
f"`openssl rand -hex 32`"
)
return key
def key_from_env(env_name: str) -> bytes | None:
"""Klucz albo None. Zły klucz to wyjątek OD RAZU — nie przy pierwszym żądaniu."""
raw = os.getenv(env_name, "")
return parse_key(raw) if raw.strip() else None
def encryption_required() -> bool:
"""Czy brak klucza ma być błędem, a nie cichym powrotem do jawnego ruchu.
Serwer sam z siebie broni się fail-closed, ale to za mało: klient BEZ klucza
wysyła pytanie otwartym tekstem i dopiero potem dostaje odmowę — czyli treść
zapytania zdążyła już przelecieć przez sieć. Ta flaga zatrzymuje go, zanim
cokolwiek opuści proces. Ustawiana razem z kluczami we wdrożeniu.
"""
return os.getenv(ENV_REQUIRED, "").strip().lower() in {"1", "true", "yes", "on"}
def _subkey(link_key: bytes, direction: bytes) -> bytes:
return HKDF(
algorithm=hashes.SHA256(), length=KEY_BYTES, salt=None,
info=b"astrololo/link/" + direction,
).derive(link_key)
class Link:
"""Jedna para rozmówców: klucz plus wyprowadzone z niego podklucze."""
def __init__(self, link_key: bytes) -> None:
self._by_direction = {
REQUEST: AESGCM(_subkey(link_key, REQUEST)),
RESPONSE: AESGCM(_subkey(link_key, RESPONSE)),
}
# ---------------------------------------------------------- pojedyncza ramka
def _aad(self, direction: bytes, path: str, stamp: str, seq: int) -> bytes:
return b"|".join([MAGIC, direction, path.encode("utf-8"),
stamp.encode("ascii"), str(seq).encode("ascii")])
def seal(self, direction: bytes, path: str, stamp: str, seq: int,
plaintext: bytes) -> bytes:
nonce = os.urandom(NONCE_BYTES)
sealed = self._by_direction[direction].encrypt(
nonce, plaintext, self._aad(direction, path, stamp, seq))
return MAGIC + nonce + sealed
def open(self, direction: bytes, path: str, stamp: str, seq: int,
frame: bytes) -> bytes:
if not frame.startswith(MAGIC):
raise LinkError("ramka bez znacznika astrololo")
body = frame[len(MAGIC):]
if len(body) <= NONCE_BYTES:
raise LinkError("ramka za krótka")
nonce, sealed = body[:NONCE_BYTES], body[NONCE_BYTES:]
try:
return self._by_direction[direction].decrypt(
nonce, sealed, self._aad(direction, path, stamp, seq))
except InvalidTag as exc:
# Jeden komunikat na wszystkie przypadki: zły klucz, podmieniony bajt,
# przeklejenie z innej ścieżki, przestawiona ramka. Rozróżnianie ich
# na zewnątrz podpowiadałoby atakującemu, w co trafił.
raise LinkError("nie udało się odszyfrować — zły klucz albo naruszone dane") from exc
# ------------------------------------------------------------ strumień ramek
def seal_stream(self, direction: bytes, path: str, stamp: str,
chunks: Iterable[bytes]) -> Iterator[bytes]:
for seq, chunk in enumerate(chunks):
yield frame_out(self.seal(direction, path, stamp, seq, chunk))
def open_stream(self, direction: bytes, path: str, stamp: str,
raw: bytes) -> Iterator[bytes]:
for seq, frame in enumerate(frames_in(raw)):
yield self.open(direction, path, stamp, seq, frame)
def open_all(self, direction: bytes, path: str, stamp: str, raw: bytes) -> bytes:
return b"".join(self.open_stream(direction, path, stamp, raw))
# ---------------------------------------------------------------- ramkowanie
def frame_out(payload: bytes) -> bytes:
return struct.pack(">I", len(payload)) + payload
def frames_in(raw: bytes) -> Iterator[bytes]:
"""Rozbiera bufor na ramki. Ucięty strumień to błąd, nie cicha strata danych."""
offset = 0
while offset < len(raw):
if offset + LENGTH_PREFIX > len(raw):
raise LinkError("urwana ramka (brak nagłówka długości)")
(size,) = struct.unpack(">I", raw[offset:offset + LENGTH_PREFIX])
if size > MAX_FRAME:
raise LinkError("ramka ponad dopuszczalny rozmiar")
offset += LENGTH_PREFIX
if offset + size > len(raw):
raise LinkError("urwana ramka (za mało danych)")
yield raw[offset:offset + size]
offset += size
def unframe_incremental(buffer: bytearray) -> Iterator[bytes]:
"""Wyjmuje z bufora KOMPLETNE ramki i zjada je; resztę zostawia na później.
Dla odbioru na żywo: kawałki przychodzą podzielone dowolnie i ramka potrafi
rozjechać się między dwa odczyty.
"""
while True:
if len(buffer) < LENGTH_PREFIX:
return
(size,) = struct.unpack(">I", buffer[:LENGTH_PREFIX])
if size > MAX_FRAME:
raise LinkError("ramka ponad dopuszczalny rozmiar")
if len(buffer) < LENGTH_PREFIX + size:
return
frame = bytes(buffer[LENGTH_PREFIX:LENGTH_PREFIX + size])
del buffer[:LENGTH_PREFIX + size]
yield frame
# ------------------------------------------------------------- świeżość ruchu
def stamp_now() -> str:
return f"{time.time():.3f}"
def check_stamp(stamp: str) -> None:
"""Odrzuca ramki spoza okna czasowego — inaczej podsłuchane żądanie dałoby się
odtworzyć w dowolnym momencie w przyszłości."""
try:
sent = float(stamp)
except (TypeError, ValueError) as exc:
raise LinkError("brak albo błędny znacznik czasu") from exc
if abs(time.time() - sent) > MAX_SKEW_SECONDS:
raise LinkError("znacznik czasu poza dopuszczalnym oknem")
# =========================================================== strona serwerowa
class LinkCryptoMiddleware:
"""Rozszyfrowuje wchodzące żądania i zaszyfrowuje wychodzące odpowiedzi.
Napisane jako czyste ASGI, nie ``@app.middleware("http")``, bo trzeba
podmienić CIAŁO żądania jeszcze zanim zobaczy je FastAPI, oraz przepuścić
odpowiedź strumieniową kawałek po kawałku, bez zbierania jej w pamięci.
"""
def __init__(self, app, link: Link | None, layer: str) -> None:
self.app = app
self.link = link
self.layer = layer
async def __call__(self, scope, receive, send):
if scope["type"] != "http" or self.link is None or scope["path"] in PUBLIC_PATHS:
return await self.app(scope, receive, send)
path = scope["path"]
headers = {k.decode("latin-1").lower(): v.decode("latin-1") for k, v in scope["headers"]}
if headers.get(HEADER_ENC.lower()) != VERSION:
# Fail-closed. Klucz jest ustawiony, więc jawne żądanie oznacza albo
# pomyłkę w konfiguracji, albo kogoś obcego — w obu wypadkach nie
# chcemy po cichu wrócić do jawnego ruchu.
log.warning("warstwa %s: odrzucone żądanie bez szyfrowania łącza (%s)",
self.layer, path)
return await _refuse(send, "Łącze międzywarstwowe wymaga szyfrowania.")
stamp = headers.get(HEADER_TS.lower(), "")
try:
check_stamp(stamp)
plaintext = self.link.open_all(REQUEST, path, stamp, await _read_body(receive))
except LinkError as exc:
log.warning("warstwa %s: %s (%s)", self.layer, exc, path)
return await _refuse(send, "Nie udało się odczytać zaszyfrowanego żądania.")
scope = dict(scope)
scope["headers"] = _rewritten_headers(scope["headers"], len(plaintext))
await self.app(scope, _replay(plaintext, receive), self._sealing_send(send, path))
def _sealing_send(self, send, path: str):
state: dict = {"stamp": "", "seq": 0}
async def sealing(message):
if message["type"] == "http.response.start":
state["stamp"] = stamp_now()
keep = [(k, v) for k, v in message.get("headers", [])
if k.lower() not in (b"content-length", b"content-type")]
message = dict(message)
message["headers"] = keep + [
(b"content-type", CONTENT_TYPE.encode()),
(HEADER_ENC.lower().encode(), VERSION.encode()),
(HEADER_TS.lower().encode(), state["stamp"].encode()),
]
return await send(message)
if message["type"] == "http.response.body":
chunk = message.get("body", b"")
sealed = b""
if chunk:
sealed = frame_out(self.link.seal(
RESPONSE, path, state["stamp"], state["seq"], chunk))
state["seq"] += 1
return await send({"type": "http.response.body", "body": sealed,
"more_body": message.get("more_body", False)})
return await send(message)
return sealing
def _rewritten_headers(raw: Iterable[tuple[bytes, bytes]], length: int):
"""Po odszyfrowaniu ciało ma inną długość i zwykły typ — inaczej FastAPI
próbowałby sparsować JSON o cudzej deklarowanej wielkości."""
kept = [(k, v) for k, v in raw if k.lower() not in (b"content-length", b"content-type")]
kept.append((b"content-length", str(length).encode()))
if length:
kept.append((b"content-type", b"application/json"))
return kept
async def _read_body(receive) -> bytes:
body = bytearray()
while True:
message = await receive()
if message["type"] == "http.disconnect":
raise LinkError("rozłączenie w trakcie odbioru żądania")
body += message.get("body", b"")
if not message.get("more_body", False):
return bytes(body)
def _replay(body: bytes, original):
"""Podstawia odszyfrowane ciało jako jedyną porcję wejścia dla aplikacji.
Po oddaniu ciała oddajemy głos ORYGINALNEMU `receive`, zamiast od razu
zgłaszać rozłączenie. Odpowiedź strumieniowa nasłuchuje bowiem rozłączenia
równolegle do wysyłania i przerywa się, gdy je zobaczy — na skróconej wersji
okno postępu dostawało pustą odpowiedź, choć zwykłe żądania działały.
"""
delivered = False
async def receive():
nonlocal delivered
if delivered:
return await original()
delivered = True
return {"type": "http.request", "body": body, "more_body": False}
return receive
async def _refuse(send, detail: str) -> None:
"""Odmowa leci JAWNIE — rozmówca właśnie pokazał, że nie umie odszyfrować,
więc zaszyfrowany komunikat o błędzie byłby dla niego nieczytelny."""
payload = f'{{"detail":"{detail}"}}'.encode("utf-8")
await send({"type": "http.response.start", "status": 400, "headers": [
(b"content-type", b"application/json"),
(b"content-length", str(len(payload)).encode()),
]})
await send({"type": "http.response.body", "body": payload})
def install(app, env_name: str, layer: str):
"""Podpina szyfrowanie łącza. Wołać PO `security.install`, żeby także odmowa
tokenowa (401) wracała zaszyfrowana — inaczej klient by jej nie odczytał."""
link_key = key_from_env(env_name)
if link_key is None and encryption_required():
# Celowo wywracamy start. Ta sama zasada co przy sekrecie logowania:
# wolimy widoczną awarię niż usługę, która wstała i po cichu nie chroni
# niczego. Pod w CrashLoop widać od razu, jawny ruch — nie.
raise LinkError(
f"{ENV_REQUIRED} jest włączone, ale {env_name} nie ustawiony — "
f"warstwa {layer} nie wystartuje bez klucza łącza"
)
if link_key is None:
log.warning(
"UWAGA: %s nie ustawiony — warstwa %s rozmawia z sąsiadem JAWNYM tekstem, "
"więc treść baz interpretacyjnych jest widoczna dla każdego, kto podsłucha "
"ruch wewnątrz sieci.", env_name, layer,
)
return None
link = Link(link_key)
app.add_middleware(LinkCryptoMiddleware, link=link, layer=layer)
log.info("warstwa %s: łącze szyfrowane (AES-256-GCM, klucz z %s)", layer, env_name)
return link
# ============================================================ strona kliencka
def call(client, method: str, url: str, *, payload=None,
headers: dict[str, str] | None = None, link: Link | None) -> bytes:
"""Żądanie do sąsiedniej warstwy; zwraca odszyfrowane ciało odpowiedzi.
Ścieżkę do materiału uwierzytelnianego bierzemy Z URL-a, a nie z osobnego
argumentu — gdyby klient i serwer liczyły ją inaczej, każde żądanie kończyłoby
się niejasnym błędem odszyfrowania.
"""
import json as _json
import httpx
request_headers = dict(headers or {})
if link is None:
if encryption_required():
# Zatrzymujemy się PRZED wysłaniem. Gdyby polecieć jawnie i dopiero
# zebrać odmowę, pytanie byłoby już na kablu — a to właśnie ono niesie
# sygnifikatory, o które pytamy bazę.
raise LinkError(
f"{ENV_REQUIRED} jest włączone, ale brak klucza łącza — żądanie "
f"NIE zostało wysłane, żeby jego treść nie poszła jawnym tekstem"
)
response = client.request(method, url, json=payload, headers=request_headers)
response.raise_for_status()
return response.content
path = httpx.URL(url).path
stamp = stamp_now()
plaintext = b"" if payload is None else _json.dumps(payload).encode("utf-8")
body = frame_out(link.seal(REQUEST, path, stamp, 0, plaintext))
request_headers.update({HEADER_ENC: VERSION, HEADER_TS: stamp,
"Content-Type": CONTENT_TYPE})
response = client.request(method, url, content=body, headers=request_headers)
if response.status_code >= 400 and response.headers.get(HEADER_ENC) != VERSION:
log.error("łącze %s odmówiło: %s", path, response.text[:200])
response.raise_for_status()
if response.headers.get(HEADER_ENC) != VERSION:
raise LinkError("odpowiedź przyszła nieszyfrowana, choć klucz łącza jest ustawiony")
reply_stamp = response.headers.get(HEADER_TS, "")
check_stamp(reply_stamp)
return link.open_all(RESPONSE, path, reply_stamp, response.content)
def call_json(client, method: str, url: str, *, payload=None,
headers: dict[str, str] | None = None, link: Link | None):
import json as _json
return _json.loads(call(client, method, url, payload=payload,
headers=headers, link=link))
def open_response_stream(response, link: Link | None) -> Iterator[bytes]:
"""Odbiór odpowiedzi płynącej kawałkami (okno postępu).
Ramka potrafi rozjechać się między dwa odczyty z gniazda, więc składamy ją
w buforze zamiast zakładać, że każdy kawałek to komplet.
"""
if link is None:
yield from response.iter_bytes()
return
if response.headers.get(HEADER_ENC) != VERSION:
raise LinkError("strumień przyszedł nieszyfrowany, choć klucz łącza jest ustawiony")
stamp = response.headers.get(HEADER_TS, "")
check_stamp(stamp)
path = response.request.url.path
buffer = bytearray()
seq = 0
for chunk in response.iter_bytes():
buffer += chunk
for frame in unframe_incremental(buffer):
yield link.open(RESPONSE, path, stamp, seq, frame)
seq += 1
if buffer:
raise LinkError("strumień urwał się w połowie ramki")
-5
View File
@@ -10,7 +10,6 @@ from contextlib import asynccontextmanager
from fastapi import FastAPI
from app import link_crypto, security
from app.config import settings
from app.models import HealthInfo, SearchQuery, SearchResult
from app.providers.factory import build_provider
@@ -25,10 +24,6 @@ async def lifespan(app: FastAPI):
app = FastAPI(title="astrololo · warstwa bazodanowa", lifespan=lifespan)
security.install(app, "danych") # token międzywarstwowy (LOG-32)
# Szyfrowanie łącza od logiki. PO `security.install`, żeby także odmowa
# tokenowa wracała zaszyfrowana — inaczej klient nie umiałby jej odczytać.
link_crypto.install(app, link_crypto.ENV_LOGIC_DATA, "danych")
@app.post("/search", response_model=SearchResult)
+1 -4
View File
@@ -18,10 +18,7 @@ class SearchQuery(BaseModel):
key: str = Field(..., description="Pole/kolumna kanoniczna, po której szukamy, np. 'name'.")
value: str = Field(..., description="Szukana wartość.")
exact: bool = Field(False, description="Dopasowanie dokładne vs. zawieranie (contains).")
# Górny limit celowo niski: to zapytanie oddaje SUROWE wiersze baz, więc wysoki
# pułap zamienia je w narzędzie do masowego pobrania (LOG-32). 5000 = tyle, ile
# realnie potrzebuje build_report na jeden obiekt.
limit: int = Field(50, ge=1, le=5000)
limit: int = Field(50, ge=1, le=50000)
fields: list[str] | None = Field(
None, description="Lista pól kanonicznych do zwrócenia; None = wszystkie."
)
-50
View File
@@ -1,50 +0,0 @@
"""Uwierzytelnianie międzywarstwowe (LOG-32).
Warstwa danych oddaje SUROWE wiersze baz — to najbardziej wrażliwy punkt całego
systemu. Bez tego kontrolera wystarczyłoby uderzyć w nią bezpośrednio, z pominięciem
i logiki, i logowania w UI. Gdy ustawiono INTERNAL_TOKEN, każde żądanie (poza /health)
musi go przynieść w nagłówku X-Astrololo-Token.
Bez INTERNAL_TOKEN kontrola jest wyłączona (dev / zgodność wstecz) — wtedy przy
starcie leci ostrzeżenie.
"""
from __future__ import annotations
import logging
import os
import secrets
from fastapi import Request
from fastapi.responses import JSONResponse
log = logging.getLogger("astrololo.security")
HEADER = "X-Astrololo-Token"
PUBLIC_PATHS = frozenset({"/health"})
def token() -> str:
"""Czytany leniwie — konfiguracja może się zmienić bez importu modułu."""
return os.getenv("INTERNAL_TOKEN", "")
def enabled() -> bool:
return bool(token())
def install(app, layer: str) -> None:
if not enabled():
log.warning(
"UWAGA: INTERNAL_TOKEN nie ustawiony — warstwa %s przyjmuje żądania od "
"kogokolwiek, kto ma do niej dostęp sieciowy.", layer,
)
@app.middleware("http")
async def _guard(request: Request, call_next):
if request.url.path in PUBLIC_PATHS or not enabled():
return await call_next(request)
got = request.headers.get(HEADER, "")
if not secrets.compare_digest(got, token()):
return JSONResponse({"detail": "Brak lub błędny token międzywarstwowy."},
status_code=401)
return await call_next(request)
-2
View File
@@ -7,5 +7,3 @@ openpyxl>=3.1
pyarrow>=18.0
SQLAlchemy>=2.0
pydantic>=2.10
# Szyfrowanie łącza między warstwami (PRE-16): AES-256-GCM + HKDF
cryptography>=44.0
+1 -25
View File
@@ -1,34 +1,10 @@
# Build wieloetapowy — bo `pyswisseph` to rozszerzenie C bez gotowych wheeli.
#
# Na PyPI (2.10.3.2) wheels kończą się na cp311 i obejmują wyłącznie i686/x86_64.
# Dla Pythona 3.12 oraz dla arm64 pip ZAWSZE kompiluje ze źródeł, a `-slim` nie ma
# kompilatora — dlatego jednoetapowy build tu padał. Kompilujemy w etapie builder,
# a do obrazu finalnego wchodzi już tylko gotowy wheel (bez toolchaina).
FROM python:3.12-slim AS builder
RUN apt-get update \
&& apt-get install -y --no-install-recommends build-essential \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /build
COPY requirements.txt .
RUN pip wheel --no-cache-dir --wheel-dir /wheels -r requirements.txt
FROM python:3.12-slim
WORKDIR /app
COPY --from=builder /wheels /wheels
COPY requirements.txt .
RUN pip install --no-cache-dir --no-index --find-links=/wheels -r requirements.txt \
&& rm -rf /wheels
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
# Sanity check na etapie budowania: brak działającego swissepha ma wywalić build,
# a nie dopiero pierwszy request.
RUN python -c "import swisseph as swe; swe.set_ephe_path(None); print('swisseph OK', swe.version)"
EXPOSE 8003
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8003"]
+1 -15
View File
@@ -15,23 +15,9 @@ permisywnej reszty systemu. **Nie wchodzi do dystrybucji zamkniętego produktu.*
Tryb Moshiera (`FLG_MOSEPH`) — bez plików efemeryd, zero konfiguracji.
## Build obrazu
```bash
docker compose --profile comparison build engine-swisseph
```
Dockerfile jest **wieloetapowy** i to nie jest ozdobnik: `pyswisseph` to rozszerzenie
C, a na PyPI (2.10.3.2) gotowe wheels kończą się na **cp311** i obejmują wyłącznie
**i686/x86_64**. Dla Pythona 3.12 oraz dla arm64 pip musi kompilować ze źródeł, więc
sam `python:3.12-slim` (bez kompilatora) build wywracał. Kompilacja idzie w etapie
`builder` (`build-essential`), a do obrazu finalnego trafia już tylko gotowy wheel —
runtime zostaje czysty i mały. Pierwszy build trwa ~12 min, kolejne idą z cache warstw.
Build kończy się sanity-checkiem (`import swisseph`), żeby niedziałający silnik
wykrzaczył build, a nie dopiero pierwszy request.
## Uruchomienie (tylko profil porównawczy / dev / CI)
```bash
pip install -r requirements.txt # wymaga kompilatora C (patrz wyżej)
pip install -r requirements.txt
uvicorn app.main:app --port 8003
```
Następnie w warstwie logicznej ustaw `ENGINE_SWISSEPH_URL=http://localhost:8003`,
+1 -1
View File
@@ -6,7 +6,7 @@ i nie w bazie.
## API
- `POST /api/query``QueryRequest``QueryResponse`
- `POST /chart/positions``{when_utc, lat, lon, house_system?}` → pełny horoskop: pozycje (LOG-01) + osie i domy (LOG-05) + aspekty główne z applying/separating (LOG-06) + opcjonalnie stacje planet (`stations:true`, LOG-03). Zwraca też sektę i 7 Lots hermetycznych z domami (LOG-08). Obiekty: 10 planet + mean NN/SN/Lilith (LOG-02). `house_system`: `whole_sign` (dom.) / `equal` / `porphyry`.
- `POST /chart/positions``{when_utc, lat, lon, house_system?}` → pełny horoskop: pozycje (LOG-01) + osie i domy (LOG-05) + aspekty główne z applying/separating (LOG-06) + opcjonalnie stacje planet (`stations:true`, LOG-03). Obiekty: 10 planet + mean NN/SN/Lilith (LOG-02). `house_system`: `whole_sign` (dom.) / `equal` / `porphyry`.
- `POST /chart/report``{when_utc, lat, lon, limit?}` → wynik obliczeń wyszukany w bazie: fasety sygnifikatorów **w znaku / w domu / w aspekcie**, z rozwinięciem skrótów, odsiewaniem duplikatów (ten sam sygnifikator i opis), rankingiem siły (LOG-21) oraz opcją group (grupowanie identycznych opisów)
- `POST /chart/profections``{when_utc, lat, lon, start_age?, count?}` → profekcje roczne: wiek, profektowany Asc, Władca Roku (+MC/Su/Mo) (LOG-10)
- `POST /chart/return``{when_utc, lat, lon, kind, around?}` → Solar/Lunar Return: moment powrotu + pełny horoskop na ten moment (LOG-12)
-2
View File
@@ -24,8 +24,6 @@ ABBREVIATIONS: dict[str, str] = {
"NN": "North Node", "SN": "South Node", "Lilith": "Lilith", "Chiron": "Chiron",
# osie
"Asc": "Ascendant", "Dsc": "Descendant", "MC": "Midheaven", "IC": "Imum Coeli",
# Lots (punkty arabskie)
"PF": "Part of Fortune", "Fortune": "Part of Fortune", "Spirit": "Lot of Spirit",
# aspekty
"conj": "conjunction", "sex": "sextile", "sq": "square", "tri": "trine",
"opp": "opposition", "semisex": "semisextile", "semisq": "semisquare",
+7 -24
View File
@@ -5,28 +5,13 @@ Jedyny punkt styku w dół. Gdyby warstwa bazodanowa zmieniła implementację
"""
from __future__ import annotations
import os
from typing import Any
import httpx
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 logika↔dane. 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_LOGIC_DATA)
return link_crypto.Link(key) if key else None
class DataClient:
def __init__(self, base_url: str | None = None) -> None:
self.base_url = (base_url or settings.data_url).rstrip("/")
@@ -41,14 +26,12 @@ class DataClient:
) -> dict[str, Any]:
payload = {"key": key, "value": value, "exact": exact, "limit": limit, "fields": fields}
with httpx.Client(timeout=max(settings.http_timeout, 30.0)) as client:
return link_crypto.call_json(client, "POST", f"{self.base_url}/search",
payload=payload, headers=_auth_headers(),
link=_link())
def health(self) -> dict[str, Any]:
# /health celowo poza szyfrowaniem — pukają tu sondy k8s, które klucza
# nie mają, a nie przechodzi tędy nic z baz.
with httpx.Client(timeout=settings.http_timeout) as client:
r = client.get(f"{self.base_url}/health", headers=_auth_headers())
r = client.post(f"{self.base_url}/search", json=payload)
r.raise_for_status()
return r.json()
def health(self) -> dict[str, Any]:
with httpx.Client(timeout=settings.http_timeout) as client:
r = client.get(f"{self.base_url}/health")
r.raise_for_status()
return r.json()
-16
View File
@@ -27,18 +27,6 @@ LUMINARIES = {"Sun", "Moon"}
DEFAULT_ORB = 8.0
LUMINARY_BONUS = 2.0
# Pary sztywno powiązane definicyjnie — kąt między nimi wynika z samej definicji
# punktu, nie z układu nieba (SN = NN + 180°). Aspekt taki zawsze wychodzi
# dokładny (orb 0,00°) i nie niesie żadnej informacji astrologicznej, więc
# wycinamy go z wyników: zaśmieca listę w UI i zjada budżet promptu do LLM.
RIGID_PAIRS = frozenset({
frozenset({"North Node", "South Node"}),
})
def _is_rigid(name_a: str, name_b: str) -> bool:
return frozenset({name_a, name_b}) in RIGID_PAIRS
def separation(a: float, b: float) -> float:
"""Najmniejsza separacja kątowa [0,180]."""
@@ -68,16 +56,12 @@ def find_aspects(
Zwraca listę aspektów głównych; gdy znane są prędkości, każdy aspekt ma
applying (bool) i skrót 'as': 'A'/'S' (aplikacyjny/separacyjny).
Pary z RIGID_PAIRS (np. NN/SN) są pomijane — ich kąt jest definicyjny.
"""
out: list[dict] = []
n = len(positions)
for i in range(n):
for j in range(i + 1, n):
a, b = positions[i], positions[j]
if _is_rigid(a["name"], b["name"]):
continue
la, lb = a.get("decimal"), b.get("decimal")
if la is None or lb is None:
continue
+11 -53
View File
@@ -7,51 +7,27 @@ pozycje.
from __future__ import annotations
from app.engine import houses as H
from app.engine import zodiac as Z
from app.engine.base import EphemerisEngine
from app.engine.formats import SIGNS, absolute, decimal, in_sign, norm360, sign_index
from app.engine.formats import SIGNS, in_sign, norm360, sign_index
from app.engine.models import ChartMoment
def _fmt(name: str, lon: float, off: float = 0.0) -> dict:
lon = norm360(lon - off)
def _fmt(name: str, lon: float) -> dict:
return {
"name": name,
"sign": SIGNS[sign_index(lon)],
"in_sign": in_sign(lon),
"decimal": round(lon, 6),
"decimal": round(norm360(lon), 6),
}
def _shift_pos(pdict: dict, off: float) -> None:
"""Przelicza etykiety pozycji na wybrany zodiak (in-place). off=0 → bez zmian."""
if not off:
return
lon = norm360(pdict["decimal"] - off)
pdict["sign"] = SIGNS[sign_index(lon)]
pdict["in_sign"] = in_sign(lon)
pdict["absolute"] = absolute(lon)
pdict["decimal"] = decimal(lon)
def build_chart(engine: EphemerisEngine, moment: ChartMoment, house_system: str = H.WHOLE_SIGN,
lots_method: str = "degree", zodiac: str = Z.TROPICAL) -> dict:
def build_chart(engine: EphemerisEngine, moment: ChartMoment, house_system: str = H.WHOLE_SIGN) -> dict:
from app.engine.aspects import find_aspects
positions = engine.positions(moment)
result: dict = {"engine": engine.name, "positions": [p.as_dict() for p in positions]}
# aspekty liczymy PRZED zmianą zodiaku — kąty między obiektami są niezmiennicze
result["aspects"] = find_aspects(result["positions"]) # aspekty (LOG-06)
# offset zodiaku (LOG-04): syderyczny = ayanamsa, draconic = długość węzła
node_lon = next((p.longitude for p in positions if p.name == "North Node"), None)
off = Z.offset(zodiac, Z.julian_day(moment.when_utc), node_lon)
result["zodiac"] = zodiac
if zodiac in Z.SIDEREAL:
result["ayanamsha"] = round(off, 6)
for pdict in result["positions"]:
_shift_pos(pdict, off)
if not hasattr(engine, "sidereal"):
return result
@@ -59,37 +35,19 @@ def build_chart(engine: EphemerisEngine, moment: ChartMoment, house_system: str
asc = H.compute_asc(ramc, eps, moment.lat)
mc = H.compute_mc(ramc, eps)
system = house_system if house_system in H.SYSTEMS else H.WHOLE_SIGN
cusp_list = H.cusps(asc, mc, system) # tropikalne — geometria domów jest niezmiennicza
cusp_list = H.cusps(asc, mc, system)
result["house_system"] = system
result["angles"] = {
"Asc": _fmt("Asc", asc, off),
"MC": _fmt("MC", mc, off),
"Dsc": _fmt("Dsc", norm360(asc + 180.0), off),
"IC": _fmt("IC", norm360(mc + 180.0), off),
"Asc": _fmt("Asc", asc),
"MC": _fmt("MC", mc),
"Dsc": _fmt("Dsc", norm360(asc + 180.0)),
"IC": _fmt("IC", norm360(mc + 180.0)),
}
result["cusps"] = [
{"house": i + 1, "sign": SIGNS[sign_index(norm360(c - off))],
"in_sign": in_sign(norm360(c - off))}
{"house": i + 1, "sign": SIGNS[sign_index(c)], "in_sign": in_sign(c)}
for i, c in enumerate(cusp_list)
]
for pdict, obj in zip(result["positions"], positions):
pdict["house"] = H.assign_house(obj.longitude, cusp_list) # dom po długości tropikalnej
# Lots (LOG-08) — wymagają Asc i sekty (dzień/noc)
from app.engine.firdaria import is_day_birth
from app.engine.lots import compute_lots
pts = {p.name: p.longitude for p in positions}
pts["Asc"] = asc
day = is_day_birth(pts["Sun"], asc, mc) if "Sun" in pts else True
result["sect"] = "day" if day else "night"
result["lots"] = [
{**lot,
"longitude": decimal(norm360(lot["longitude"] - off)), # w wybranym zodiaku
"sign": SIGNS[sign_index(norm360(lot["longitude"] - off))],
"in_sign": in_sign(norm360(lot["longitude"] - off)),
"house": H.assign_house(lot["longitude"], cusp_list)} # dom po długości tropikalnej
for lot in compute_lots(pts, day, lots_method)
]
pdict["house"] = H.assign_house(obj.longitude, cusp_list)
return result
-60
View File
@@ -1,60 +0,0 @@
"""Lots / punkty arabskie (LOG-08) — 7 Lots hermetycznych.
Formuła: Lot = C + A B (od punktu C odmierzamy odległość między A i B).
Większość Lots **odwraca się w horoskopach nocnych** (zamiana A↔B) — np.
Fortuna: dzień Asc + Mo Su, noc Asc + Su Mo.
Dwa warianty liczenia (notes3):
- `degree` (domyślny) — dokładny stopień,
- `sign` — liczone całymi znakami (Lot wypada na 0° wyliczonego znaku).
Kolejność ma znaczenie: Fortuna i Duch liczone są pierwsze, bo pozostałe Lots
odwołują się do nich.
"""
from __future__ import annotations
from app.engine.formats import norm360, sign_index
# (nazwa, C, A, B, odwracalny w nocy)
LOT_DEFS: list[tuple[str, str, str, str, bool]] = [
("Fortune", "Asc", "Moon", "Sun", True),
("Spirit", "Asc", "Sun", "Moon", True),
("Eros", "Asc", "Venus", "Spirit", True),
("Necessity", "Asc", "Fortune", "Mercury", True),
("Courage", "Asc", "Fortune", "Mars", True),
("Victory", "Asc", "Jupiter", "Spirit", True),
("Nemesis", "Asc", "Fortune", "Saturn", True),
]
METHODS = ("degree", "sign")
def compute_lots(
points: dict[str, float], is_day: bool, method: str = "degree"
) -> list[dict]:
"""points: nazwa → długość natalna (wymagane Asc + planety formuł).
Zwraca listę {name, longitude, formula} w kolejności definicji.
"""
if method not in METHODS:
raise ValueError(f"nieznana metoda liczenia Lots: {method}")
vals = dict(points)
out: list[dict] = []
for name, c, a, b, reversible in LOT_DEFS:
first, second = (a, b) if (is_day or not reversible) else (b, a)
if any(k not in vals for k in (c, first, second)):
continue # brak składnika — pomijamy
if method == "sign":
idx = (sign_index(vals[c]) + sign_index(vals[first])
- sign_index(vals[second])) % 12
lon = idx * 30.0
else:
lon = norm360(vals[c] + vals[first] - vals[second])
vals[name] = lon # dostępny dla kolejnych Lots
out.append({
"name": name,
"longitude": lon,
"formula": f"{c} + {first} {second}",
})
return out
-103
View File
@@ -1,103 +0,0 @@
"""Systemy zodiaku (LOG-04): tropikalny, syderyczny (ayanamsy), draconic + RA.
Wszystkie pozycje silnika są liczone **tropikalnie of-date** (kontrakt LOG-28).
Zmiana zodiaku to — dla zodiaków ekliptycznych — jednolite przesunięcie długości:
długość_docelowa = (długość_tropikalna offset) mod 360
gdzie offset to:
- **syderyczny**: ayanamsa (kąt między tropikalnym a syderycznym punktem Barana),
- **draconic**: długość wznoszącego węzła Księżyca (węzeł = 0° draconic),
- **tropikalny**: 0.
Ponieważ to stałe przesunięcie obiektu ORAZ cusps, **numery domów się nie zmieniają**
(geometria jest niezmiennicza względem obrotu) — przesuwamy tylko etykiety znaków.
Model ayanamsy: `ayan(jd) = ayan0 + B·x + C·x²`, gdzie `x = jd J2000`. Prędkość
precesji (B, C) jest **wspólna** dla wszystkich ayanams; różni je tylko stała `ayan0`
(wybór syderycznego zera). Stałe skalibrowano do Swiss Ephemeris jako wyroczni —
zgodność do ~0,02" w latach 19002100 (patrz tests/test_zodiac.py).
RA (right ascension): konwersja ekliptyka→równik dla przyszłego widoku równikowego.
"""
from __future__ import annotations
import math
from datetime import datetime
from app.engine.formats import norm360
TROPICAL = "tropical"
DRACONIC = "draconic"
# stała ayanamsy w J2000.0 (°) — skalibrowana do swisseph (get_ayanamsa_ut)
_AYAN0 = {
"lahiri": 23.857092,
"fagan_bradley": 24.740300,
"krishnamurti": 23.760240,
}
_J2000 = 2451545.0
_B = 3.824459e-5 # °/dobę — liniowy człon precesji (wspólny)
_C = 2.304e-13 # °/dobę² — drobne przyspieszenie (wspólne)
# nazwy zodiaków akceptowane przez API
SIDEREAL = tuple(f"sidereal_{k}" for k in _AYAN0) # sidereal_lahiri, ...
SYSTEMS = (TROPICAL, *SIDEREAL, DRACONIC)
def julian_day(dt: datetime) -> float:
"""Julian Day (UT) z momentu UTC — algorytm Meeusa (kalendarz gregoriański)."""
y, m = dt.year, dt.month
day = dt.day + (dt.hour + dt.minute / 60.0 + dt.second / 3600.0
+ dt.microsecond / 3.6e9) / 24.0
if m <= 2:
y -= 1
m += 12
a = y // 100
b = 2 - a + a // 4
return math.floor(365.25 * (y + 4716)) + math.floor(30.6001 * (m + 1)) + day + b - 1524.5
def ayanamsha(name: str, jd: float) -> float:
"""Ayanamsa [°] danej szkoły dla Julian Day (UT)."""
key = name[len("sidereal_"):] if name.startswith("sidereal_") else name
if key not in _AYAN0:
raise ValueError(f"Nieznana ayanamsa: {name!r} (dostępne: {', '.join(_AYAN0)})")
x = jd - _J2000
return _AYAN0[key] + _B * x + _C * x * x
def offset(zodiac: str, jd: float, node_lon: float | None = None) -> float:
"""Ile odjąć od długości tropikalnej, by dostać wybrany zodiak.
`node_lon` (tropikalna długość węzła wznoszącego) wymagana tylko dla draconic.
"""
if zodiac == TROPICAL:
return 0.0
if zodiac == DRACONIC:
if node_lon is None:
raise ValueError("draconic wymaga długości węzła (node_lon)")
return norm360(node_lon)
if zodiac in SIDEREAL:
return ayanamsha(zodiac, jd)
raise ValueError(f"Nieznany zodiak: {zodiac!r} (dostępne: {', '.join(SYSTEMS)})")
def apply(lon: float, off: float) -> float:
"""Długość w docelowym zodiaku."""
return norm360(lon - off)
def to_equatorial(lon: float, lat: float, eps: float) -> tuple[float, float]:
"""Ekliptyka (λ, β) → równik: (RA, deklinacja) w stopniach. Wszystko w °.
RA rośnie 0360°; deklinacja w [90, 90].
"""
lam, bet, e = math.radians(lon), math.radians(lat), math.radians(eps)
sin_dec = math.sin(bet) * math.cos(e) + math.cos(bet) * math.sin(e) * math.sin(lam)
dec = math.asin(max(-1.0, min(1.0, sin_dec)))
ra = math.atan2(
math.sin(lam) * math.cos(e) - math.tan(bet) * math.sin(e),
math.cos(lam),
)
return norm360(math.degrees(ra)), math.degrees(dec)
-466
View File
@@ -1,466 +0,0 @@
"""Szyfrowanie łączy między warstwami (PRE-16 / LOG-33).
Do tej pory warstwy rozmawiały ze sobą zwykłym HTTP-em wewnątrz klastra. Token
międzywarstwowy (LOG-32) mówił KTO pyta, ale nie ukrywał CZEGO dotyczy odpowiedź
— a płyną nią surowe wiersze oryginalnych baz interpretacyjnych, czyli rdzeń
produktu. Kto podsłuchał ruch wewnątrz sieci (drugi pod, port mirror na switchu,
zrzut z węzła), miał je w całości.
Ten moduł zamyka tę drogę: **AES-256-GCM** na ciele każdego żądania i odpowiedzi.
GCM daje jednocześnie poufność i uwierzytelnienie — cudzy albo podmieniony bajt
nie odszyfruje się w ogóle, więc nie ma osobnego problemu „zaszyfrowane, ale
podatne na modyfikację".
**Dwa niezależne klucze**, po jednym na parę rozmówców:
* ``LINK_KEY_PRESENTATION_LOGIC`` — prezentacja ↔ logika,
* ``LINK_KEY_LOGIC_DATA`` — logika ↔ dane.
Dzięki temu przejęcie klucza prezentacji nie daje dostępu do warstwy danych,
gdzie leżą całe bazy. Logika trzyma oba, bo rozmawia w obie strony.
Z każdego klucza łącza wyprowadzamy **osobne podklucze na kierunek** (HKDF).
Żądanie i odpowiedź nigdy nie szyfrują się tym samym kluczem, więc powtórzenie
losowej jednorazówki w jedną stronę nie osłabia drugiej.
Format ramki (bo strumień odpowiedzi może iść kawałkami — patrz okno postępu):
[4 bajty długości][magia "AL1"][12 bajtów jednorazówki][szyfrogram + znacznik]
Do materiału uwierzytelnianego (AAD) wchodzą kierunek, ścieżka, znacznik czasu
i numer ramki. Skutek: ramki nie da się przekleić do innego endpointu, odtworzyć
po czasie (dopuszczalny poślizg ``MAX_SKEW``) ani przestawić w strumieniu.
Bez ustawionego klucza moduł **przepuszcza ruch otwartym tekstem** (dev, zgodność
wstecz) i krzyczy o tym przy starcie. Gdy klucz JEST ustawiony, warstwa serwerowa
działa fail-closed: nieszyfrowane żądanie dostaje odmowę, żeby przypadkowa
regresja po stronie klienta nie oznaczała cichego powrotu do jawnego ruchu.
"""
from __future__ import annotations
import base64
import binascii
import logging
import os
import struct
import time
from typing import Iterable, Iterator
from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
log = logging.getLogger("astrololo.link")
MAGIC = b"AL1"
VERSION = "v1"
NONCE_BYTES = 12
KEY_BYTES = 32 # AES-256
LENGTH_PREFIX = 4
MAX_FRAME = 64 * 1024 * 1024 # zapora przed alokacją z podanej długości
MAX_SKEW_SECONDS = 300.0
HEADER_ENC = "X-Astrololo-Enc"
HEADER_TS = "X-Astrololo-Enc-Ts"
CONTENT_TYPE = "application/vnd.astrololo.enc"
ENV_PRESENTATION_LOGIC = "LINK_KEY_PRESENTATION_LOGIC"
ENV_LOGIC_DATA = "LINK_KEY_LOGIC_DATA"
ENV_REQUIRED = "LINK_ENCRYPTION_REQUIRED"
REQUEST, RESPONSE = b"req", b"res"
# Sondy k8s pukają tu bez klucza i tak ma zostać — inaczej pierwsza literówka
# w sekrecie kładłaby pody zamiast pokazać błąd w aplikacji.
PUBLIC_PATHS = frozenset({"/health"})
class LinkError(Exception):
"""Cokolwiek poszło nie tak z kopertą — celowo bez szczegółów na zewnątrz."""
# --------------------------------------------------------------------- klucze
def parse_key(raw: str) -> bytes:
"""Klucz z konfiguracji: hex (64 znaki) albo base64. Zawsze 32 bajty."""
text = raw.strip()
if not text:
raise LinkError("pusty klucz łącza")
try:
key = bytes.fromhex(text)
except ValueError:
try:
key = base64.b64decode(text, validate=True)
except (binascii.Error, ValueError) as exc:
raise LinkError("klucz łącza nie jest ani hexem, ani base64") from exc
if len(key) != KEY_BYTES:
raise LinkError(
f"klucz łącza ma {len(key)} B zamiast {KEY_BYTES} — wygeneruj przez "
f"`openssl rand -hex 32`"
)
return key
def key_from_env(env_name: str) -> bytes | None:
"""Klucz albo None. Zły klucz to wyjątek OD RAZU — nie przy pierwszym żądaniu."""
raw = os.getenv(env_name, "")
return parse_key(raw) if raw.strip() else None
def encryption_required() -> bool:
"""Czy brak klucza ma być błędem, a nie cichym powrotem do jawnego ruchu.
Serwer sam z siebie broni się fail-closed, ale to za mało: klient BEZ klucza
wysyła pytanie otwartym tekstem i dopiero potem dostaje odmowę — czyli treść
zapytania zdążyła już przelecieć przez sieć. Ta flaga zatrzymuje go, zanim
cokolwiek opuści proces. Ustawiana razem z kluczami we wdrożeniu.
"""
return os.getenv(ENV_REQUIRED, "").strip().lower() in {"1", "true", "yes", "on"}
def _subkey(link_key: bytes, direction: bytes) -> bytes:
return HKDF(
algorithm=hashes.SHA256(), length=KEY_BYTES, salt=None,
info=b"astrololo/link/" + direction,
).derive(link_key)
class Link:
"""Jedna para rozmówców: klucz plus wyprowadzone z niego podklucze."""
def __init__(self, link_key: bytes) -> None:
self._by_direction = {
REQUEST: AESGCM(_subkey(link_key, REQUEST)),
RESPONSE: AESGCM(_subkey(link_key, RESPONSE)),
}
# ---------------------------------------------------------- pojedyncza ramka
def _aad(self, direction: bytes, path: str, stamp: str, seq: int) -> bytes:
return b"|".join([MAGIC, direction, path.encode("utf-8"),
stamp.encode("ascii"), str(seq).encode("ascii")])
def seal(self, direction: bytes, path: str, stamp: str, seq: int,
plaintext: bytes) -> bytes:
nonce = os.urandom(NONCE_BYTES)
sealed = self._by_direction[direction].encrypt(
nonce, plaintext, self._aad(direction, path, stamp, seq))
return MAGIC + nonce + sealed
def open(self, direction: bytes, path: str, stamp: str, seq: int,
frame: bytes) -> bytes:
if not frame.startswith(MAGIC):
raise LinkError("ramka bez znacznika astrololo")
body = frame[len(MAGIC):]
if len(body) <= NONCE_BYTES:
raise LinkError("ramka za krótka")
nonce, sealed = body[:NONCE_BYTES], body[NONCE_BYTES:]
try:
return self._by_direction[direction].decrypt(
nonce, sealed, self._aad(direction, path, stamp, seq))
except InvalidTag as exc:
# Jeden komunikat na wszystkie przypadki: zły klucz, podmieniony bajt,
# przeklejenie z innej ścieżki, przestawiona ramka. Rozróżnianie ich
# na zewnątrz podpowiadałoby atakującemu, w co trafił.
raise LinkError("nie udało się odszyfrować — zły klucz albo naruszone dane") from exc
# ------------------------------------------------------------ strumień ramek
def seal_stream(self, direction: bytes, path: str, stamp: str,
chunks: Iterable[bytes]) -> Iterator[bytes]:
for seq, chunk in enumerate(chunks):
yield frame_out(self.seal(direction, path, stamp, seq, chunk))
def open_stream(self, direction: bytes, path: str, stamp: str,
raw: bytes) -> Iterator[bytes]:
for seq, frame in enumerate(frames_in(raw)):
yield self.open(direction, path, stamp, seq, frame)
def open_all(self, direction: bytes, path: str, stamp: str, raw: bytes) -> bytes:
return b"".join(self.open_stream(direction, path, stamp, raw))
# ---------------------------------------------------------------- ramkowanie
def frame_out(payload: bytes) -> bytes:
return struct.pack(">I", len(payload)) + payload
def frames_in(raw: bytes) -> Iterator[bytes]:
"""Rozbiera bufor na ramki. Ucięty strumień to błąd, nie cicha strata danych."""
offset = 0
while offset < len(raw):
if offset + LENGTH_PREFIX > len(raw):
raise LinkError("urwana ramka (brak nagłówka długości)")
(size,) = struct.unpack(">I", raw[offset:offset + LENGTH_PREFIX])
if size > MAX_FRAME:
raise LinkError("ramka ponad dopuszczalny rozmiar")
offset += LENGTH_PREFIX
if offset + size > len(raw):
raise LinkError("urwana ramka (za mało danych)")
yield raw[offset:offset + size]
offset += size
def unframe_incremental(buffer: bytearray) -> Iterator[bytes]:
"""Wyjmuje z bufora KOMPLETNE ramki i zjada je; resztę zostawia na później.
Dla odbioru na żywo: kawałki przychodzą podzielone dowolnie i ramka potrafi
rozjechać się między dwa odczyty.
"""
while True:
if len(buffer) < LENGTH_PREFIX:
return
(size,) = struct.unpack(">I", buffer[:LENGTH_PREFIX])
if size > MAX_FRAME:
raise LinkError("ramka ponad dopuszczalny rozmiar")
if len(buffer) < LENGTH_PREFIX + size:
return
frame = bytes(buffer[LENGTH_PREFIX:LENGTH_PREFIX + size])
del buffer[:LENGTH_PREFIX + size]
yield frame
# ------------------------------------------------------------- świeżość ruchu
def stamp_now() -> str:
return f"{time.time():.3f}"
def check_stamp(stamp: str) -> None:
"""Odrzuca ramki spoza okna czasowego — inaczej podsłuchane żądanie dałoby się
odtworzyć w dowolnym momencie w przyszłości."""
try:
sent = float(stamp)
except (TypeError, ValueError) as exc:
raise LinkError("brak albo błędny znacznik czasu") from exc
if abs(time.time() - sent) > MAX_SKEW_SECONDS:
raise LinkError("znacznik czasu poza dopuszczalnym oknem")
# =========================================================== strona serwerowa
class LinkCryptoMiddleware:
"""Rozszyfrowuje wchodzące żądania i zaszyfrowuje wychodzące odpowiedzi.
Napisane jako czyste ASGI, nie ``@app.middleware("http")``, bo trzeba
podmienić CIAŁO żądania jeszcze zanim zobaczy je FastAPI, oraz przepuścić
odpowiedź strumieniową kawałek po kawałku, bez zbierania jej w pamięci.
"""
def __init__(self, app, link: Link | None, layer: str) -> None:
self.app = app
self.link = link
self.layer = layer
async def __call__(self, scope, receive, send):
if scope["type"] != "http" or self.link is None or scope["path"] in PUBLIC_PATHS:
return await self.app(scope, receive, send)
path = scope["path"]
headers = {k.decode("latin-1").lower(): v.decode("latin-1") for k, v in scope["headers"]}
if headers.get(HEADER_ENC.lower()) != VERSION:
# Fail-closed. Klucz jest ustawiony, więc jawne żądanie oznacza albo
# pomyłkę w konfiguracji, albo kogoś obcego — w obu wypadkach nie
# chcemy po cichu wrócić do jawnego ruchu.
log.warning("warstwa %s: odrzucone żądanie bez szyfrowania łącza (%s)",
self.layer, path)
return await _refuse(send, "Łącze międzywarstwowe wymaga szyfrowania.")
stamp = headers.get(HEADER_TS.lower(), "")
try:
check_stamp(stamp)
plaintext = self.link.open_all(REQUEST, path, stamp, await _read_body(receive))
except LinkError as exc:
log.warning("warstwa %s: %s (%s)", self.layer, exc, path)
return await _refuse(send, "Nie udało się odczytać zaszyfrowanego żądania.")
scope = dict(scope)
scope["headers"] = _rewritten_headers(scope["headers"], len(plaintext))
await self.app(scope, _replay(plaintext, receive), self._sealing_send(send, path))
def _sealing_send(self, send, path: str):
state: dict = {"stamp": "", "seq": 0}
async def sealing(message):
if message["type"] == "http.response.start":
state["stamp"] = stamp_now()
keep = [(k, v) for k, v in message.get("headers", [])
if k.lower() not in (b"content-length", b"content-type")]
message = dict(message)
message["headers"] = keep + [
(b"content-type", CONTENT_TYPE.encode()),
(HEADER_ENC.lower().encode(), VERSION.encode()),
(HEADER_TS.lower().encode(), state["stamp"].encode()),
]
return await send(message)
if message["type"] == "http.response.body":
chunk = message.get("body", b"")
sealed = b""
if chunk:
sealed = frame_out(self.link.seal(
RESPONSE, path, state["stamp"], state["seq"], chunk))
state["seq"] += 1
return await send({"type": "http.response.body", "body": sealed,
"more_body": message.get("more_body", False)})
return await send(message)
return sealing
def _rewritten_headers(raw: Iterable[tuple[bytes, bytes]], length: int):
"""Po odszyfrowaniu ciało ma inną długość i zwykły typ — inaczej FastAPI
próbowałby sparsować JSON o cudzej deklarowanej wielkości."""
kept = [(k, v) for k, v in raw if k.lower() not in (b"content-length", b"content-type")]
kept.append((b"content-length", str(length).encode()))
if length:
kept.append((b"content-type", b"application/json"))
return kept
async def _read_body(receive) -> bytes:
body = bytearray()
while True:
message = await receive()
if message["type"] == "http.disconnect":
raise LinkError("rozłączenie w trakcie odbioru żądania")
body += message.get("body", b"")
if not message.get("more_body", False):
return bytes(body)
def _replay(body: bytes, original):
"""Podstawia odszyfrowane ciało jako jedyną porcję wejścia dla aplikacji.
Po oddaniu ciała oddajemy głos ORYGINALNEMU `receive`, zamiast od razu
zgłaszać rozłączenie. Odpowiedź strumieniowa nasłuchuje bowiem rozłączenia
równolegle do wysyłania i przerywa się, gdy je zobaczy — na skróconej wersji
okno postępu dostawało pustą odpowiedź, choć zwykłe żądania działały.
"""
delivered = False
async def receive():
nonlocal delivered
if delivered:
return await original()
delivered = True
return {"type": "http.request", "body": body, "more_body": False}
return receive
async def _refuse(send, detail: str) -> None:
"""Odmowa leci JAWNIE — rozmówca właśnie pokazał, że nie umie odszyfrować,
więc zaszyfrowany komunikat o błędzie byłby dla niego nieczytelny."""
payload = f'{{"detail":"{detail}"}}'.encode("utf-8")
await send({"type": "http.response.start", "status": 400, "headers": [
(b"content-type", b"application/json"),
(b"content-length", str(len(payload)).encode()),
]})
await send({"type": "http.response.body", "body": payload})
def install(app, env_name: str, layer: str):
"""Podpina szyfrowanie łącza. Wołać PO `security.install`, żeby także odmowa
tokenowa (401) wracała zaszyfrowana — inaczej klient by jej nie odczytał."""
link_key = key_from_env(env_name)
if link_key is None and encryption_required():
# Celowo wywracamy start. Ta sama zasada co przy sekrecie logowania:
# wolimy widoczną awarię niż usługę, która wstała i po cichu nie chroni
# niczego. Pod w CrashLoop widać od razu, jawny ruch — nie.
raise LinkError(
f"{ENV_REQUIRED} jest włączone, ale {env_name} nie ustawiony — "
f"warstwa {layer} nie wystartuje bez klucza łącza"
)
if link_key is None:
log.warning(
"UWAGA: %s nie ustawiony — warstwa %s rozmawia z sąsiadem JAWNYM tekstem, "
"więc treść baz interpretacyjnych jest widoczna dla każdego, kto podsłucha "
"ruch wewnątrz sieci.", env_name, layer,
)
return None
link = Link(link_key)
app.add_middleware(LinkCryptoMiddleware, link=link, layer=layer)
log.info("warstwa %s: łącze szyfrowane (AES-256-GCM, klucz z %s)", layer, env_name)
return link
# ============================================================ strona kliencka
def call(client, method: str, url: str, *, payload=None,
headers: dict[str, str] | None = None, link: Link | None) -> bytes:
"""Żądanie do sąsiedniej warstwy; zwraca odszyfrowane ciało odpowiedzi.
Ścieżkę do materiału uwierzytelnianego bierzemy Z URL-a, a nie z osobnego
argumentu — gdyby klient i serwer liczyły ją inaczej, każde żądanie kończyłoby
się niejasnym błędem odszyfrowania.
"""
import json as _json
import httpx
request_headers = dict(headers or {})
if link is None:
if encryption_required():
# Zatrzymujemy się PRZED wysłaniem. Gdyby polecieć jawnie i dopiero
# zebrać odmowę, pytanie byłoby już na kablu — a to właśnie ono niesie
# sygnifikatory, o które pytamy bazę.
raise LinkError(
f"{ENV_REQUIRED} jest włączone, ale brak klucza łącza — żądanie "
f"NIE zostało wysłane, żeby jego treść nie poszła jawnym tekstem"
)
response = client.request(method, url, json=payload, headers=request_headers)
response.raise_for_status()
return response.content
path = httpx.URL(url).path
stamp = stamp_now()
plaintext = b"" if payload is None else _json.dumps(payload).encode("utf-8")
body = frame_out(link.seal(REQUEST, path, stamp, 0, plaintext))
request_headers.update({HEADER_ENC: VERSION, HEADER_TS: stamp,
"Content-Type": CONTENT_TYPE})
response = client.request(method, url, content=body, headers=request_headers)
if response.status_code >= 400 and response.headers.get(HEADER_ENC) != VERSION:
log.error("łącze %s odmówiło: %s", path, response.text[:200])
response.raise_for_status()
if response.headers.get(HEADER_ENC) != VERSION:
raise LinkError("odpowiedź przyszła nieszyfrowana, choć klucz łącza jest ustawiony")
reply_stamp = response.headers.get(HEADER_TS, "")
check_stamp(reply_stamp)
return link.open_all(RESPONSE, path, reply_stamp, response.content)
def call_json(client, method: str, url: str, *, payload=None,
headers: dict[str, str] | None = None, link: Link | None):
import json as _json
return _json.loads(call(client, method, url, payload=payload,
headers=headers, link=link))
def open_response_stream(response, link: Link | None) -> Iterator[bytes]:
"""Odbiór odpowiedzi płynącej kawałkami (okno postępu).
Ramka potrafi rozjechać się między dwa odczyty z gniazda, więc składamy ją
w buforze zamiast zakładać, że każdy kawałek to komplet.
"""
if link is None:
yield from response.iter_bytes()
return
if response.headers.get(HEADER_ENC) != VERSION:
raise LinkError("strumień przyszedł nieszyfrowany, choć klucz łącza jest ustawiony")
stamp = response.headers.get(HEADER_TS, "")
check_stamp(stamp)
path = response.request.url.path
buffer = bytearray()
seq = 0
for chunk in response.iter_bytes():
buffer += chunk
for frame in unframe_incremental(buffer):
yield link.open(RESPONSE, path, stamp, seq, frame)
seq += 1
if buffer:
raise LinkError("strumień urwał się w połowie ramki")
-1
View File
@@ -1 +0,0 @@
"""Warstwa dostawców modeli językowych (LOG-31)."""
-42
View File
@@ -1,42 +0,0 @@
"""Kontrakt dostawcy modelu językowego (LOG-31).
Analogicznie do `EphemerisEngine` (LOG-24): prezentacja i reszta logiki nie wiedzą,
kto pisze tekst — lokalny model na naszym sprzęcie czy dostawca w chmurze.
Kluczowa własność dla bezpieczeństwa baz (LOG-32): każdy dostawca deklaruje
`leaves_lan`. Prompt niesie ORYGINALNE opisy z baz, więc interfejs musi jawnie
mówić, czy ta treść opuszcza naszą sieć — UI ma na tej podstawie ostrzegać.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
@dataclass
class Completion:
"""Wynik generowania — tekst + metryki do rozliczenia i podglądu."""
text: str
model: str
provider: str
leaves_lan: bool
usage: dict = field(default_factory=dict) # prompt_tokens / completion_tokens
class LLMError(RuntimeError):
"""Błąd wołania modelu — z komunikatem nadającym się do pokazania użytkownikowi."""
class LLMProvider(ABC):
name: str = "?"
#: czy treść promptu (a więc opisy z baz) opuszcza naszą sieć
leaves_lan: bool = True
@abstractmethod
def generate(self, prompt: str, max_tokens: int) -> Completion:
"""Zwraca gotowy tekst. Rzuca LLMError przy niepowodzeniu."""
@abstractmethod
def health(self) -> dict:
"""Czy dostawca jest osiągalny i skonfigurowany."""
-75
View File
@@ -1,75 +0,0 @@
"""Katalog modeli do wyboru w UI (LOG-31).
To są **podpowiedzi**, nie zamknięta lista. Pole modelu w UI jest tekstowe z
datalistą, więc można wpisać dowolny identyfikator — konto może mieć dostęp do
modeli, których tu nie ma, a nowe wychodzą szybciej, niż aktualizuje się kod.
Puste pole = model domyślny dostawcy.
Uwaga o pewności danych:
* modele **Anthropic** pochodzą z oficjalnej dokumentacji API (okna kontekstu
i limity wyjścia zgadzają się z `app/llm/limits.py`);
* modele **OpenAI** to podpowiedzi — nie weryfikowałem ich katalogu, więc
traktuj je jako wygodę, a nie źródło prawdy;
* modele **lokalne** zależą wyłącznie od tego, co masz pobrane w Ollamie/vLLM.
Katalog można nadpisać/rozszerzyć zmienną `<DOSTAWCA>_MODELS` (lista po przecinku),
np. `OPENAI_MODELS="gpt-5,gpt-4o"`.
"""
from __future__ import annotations
import os
from app.llm.limits import limits_for
# dostawca -> [(id modelu, krótki opis dla człowieka)]
_CATALOG: dict[str, list[tuple[str, str]]] = {
"anthropic": [
("claude-opus-4-8", "Opus 4.8 — domyślny, bardzo zdolny, 1M kontekstu"),
("claude-fable-5", "Fable 5 — najbardziej zdolny, do najtrudniejszych zadań"),
("claude-sonnet-5", "Sonnet 5 — szybszy i tańszy, jakość blisko Opusa"),
("claude-opus-4-7", "Opus 4.7 — poprzednia generacja Opusa"),
("claude-haiku-4-5", "Haiku 4.5 — najszybszy i najtańszy, mniejsze okno"),
],
"openai": [
("gpt-4o-mini", "GPT-4o mini — tani i szybki"),
("gpt-4o", "GPT-4o"),
("gpt-5", "GPT-5 — jeśli Twoje konto ma dostęp"),
("gpt-4.1", "GPT-4.1"),
("gpt-4.1-mini", "GPT-4.1 mini"),
],
"local": [
("llama3.1:8b", "Llama 3.1 8B"),
("llama3.2", "Llama 3.2"),
("qwen2.5", "Qwen 2.5 — większe okno kontekstu"),
("mistral", "Mistral"),
],
}
def models_for(provider: str) -> list[dict]:
"""Podpowiedzi modeli dla dostawcy, wraz z oknem kontekstu.
Okno kontekstu podajemy, bo wprost przekłada się na opcję „maksymalny
kontekst modelu" — użytkownik widzi, na ile budżetu promptu może liczyć.
"""
override = os.getenv(f"{provider.upper()}_MODELS", "").strip()
if override:
entries = [(m.strip(), "") for m in override.split(",") if m.strip()]
else:
entries = _CATALOG.get(provider, [])
out = []
for model_id, label in entries:
context_window, max_output = limits_for(provider, model_id)
out.append({
"id": model_id,
"label": label or model_id,
"context_window": context_window,
"max_output": max_output,
})
return out
def catalog() -> dict[str, list[dict]]:
"""Pełny katalog dla UI — jedno żądanie zamiast trzech."""
return {provider: models_for(provider) for provider in ("local", "anthropic", "openai")}
-121
View File
@@ -1,121 +0,0 @@
"""Wybór dostawcy LLM (LOG-31) — jedyne miejsce znające konkretne implementacje.
Domyślny jest **model lokalny**: prompt niesie oryginalne opisy z baz, więc
domyślnie nic nie opuszcza naszej sieci (LOG-32). Chmurę włącza się świadomie —
przez konfigurację albo pojedyncze żądanie.
Konfiguracja jest **per dostawca**, bo UI pozwala przełączać go przy każdym żądaniu.
Wspólne `LLM_*` nie wystarczy: ustawienie `LLM_BASE_URL` na lokalny model kierowałoby
tam także żądania do OpenAI, a `LLM_MODEL=llama3.1:8b` kazałoby Anthropic użyć modelu
llama. Dlatego każdy dostawca ma własny komplet zmiennych.
Zmienne środowiskowe:
LLM_PROVIDER local (domyślnie) | openai | anthropic — dostawca domyślny
LLM_TIMEOUT sekundy (domyślnie 120)
LLM_MAX_TOKENS limit długości odpowiedzi (domyślnie 2000)
<DOSTAWCA>_MODEL / _BASE_URL / _API_KEY — konfiguracja konkretnego dostawcy:
LOCAL_MODEL, LOCAL_BASE_URL (klucz zwykle zbędny)
OPENAI_MODEL, OPENAI_BASE_URL, OPENAI_API_KEY
ANTHROPIC_MODEL, ANTHROPIC_BASE_URL, ANTHROPIC_API_KEY
Klucze WYŁĄCZNIE z sekretu — nigdy w repo, w UI ani w logach.
Zgodność wstecz: wspólne `LLM_MODEL` / `LLM_BASE_URL` / `LLM_API_KEY` nadal działają,
ale stosują się TYLKO do dostawcy domyślnego (LLM_PROVIDER) — czyli konfiguracja
instalacji jednodostawcowej zostaje nietknięta, a pozostali dostawcy jej nie dziedziczą.
"""
from __future__ import annotations
import os
from app.llm.base import LLMError, LLMProvider
from app.llm.providers import AnthropicProvider, ChatCompletionsProvider
LOCAL = "local"
OPENAI = "openai"
ANTHROPIC = "anthropic"
PROVIDERS = (LOCAL, OPENAI, ANTHROPIC)
_DEFAULT_MODEL = {
LOCAL: "llama3.1:8b",
OPENAI: "gpt-4o-mini",
# Opus 4.8 świadomie zamiast Sonnet 5: Sonnet uruchamia myślenie adaptacyjne,
# gdy pominąć parametr `thinking`, a jego tokeny liczą się do max_tokens —
# przy ciasnym limicie cała tura wychodziła jako samo myślenie z pustym
# tekstem. To była przyczyna pustych odpowiedzi na Anthropicu.
ANTHROPIC: "claude-opus-4-8",
}
_DEFAULT_URL = {
# Ollama i vLLM wystawiają zgodne API pod /v1
LOCAL: "http://localhost:11434/v1",
OPENAI: "https://api.openai.com/v1",
ANTHROPIC: "https://api.anthropic.com",
}
def default_provider_name() -> str:
return os.getenv("LLM_PROVIDER", LOCAL).lower()
def max_tokens() -> int:
return int(os.getenv("LLM_MAX_TOKENS", "2000"))
def timeout() -> float:
return float(os.getenv("LLM_TIMEOUT", "120"))
def setting(provider: str, suffix: str, fallback: str = "") -> str:
"""Ustawienie dostawcy: <DOSTAWCA>_<SUFIKS> → LLM_<SUFIKS> → wbudowana domyślna.
Wspólne `LLM_*` stosuje się WYŁĄCZNIE do dostawcy domyślnego — inaczej adres
lokalnego modelu przejąłby żądania do chmury (i odwrotnie).
"""
specific = os.getenv(f"{provider.upper()}_{suffix}")
if specific:
return specific
if provider == default_provider_name():
generic = os.getenv(f"LLM_{suffix}")
if generic:
return generic
return fallback
def resolve_model(name: str | None = None, model: str | None = None) -> tuple[str, str]:
"""(dostawca, model) BEZ budowania dostawcy — czyli bez wymogu klucza API.
Rozmiar budżetu promptu zależy tylko od okna kontekstu modelu, więc nie może
zależeć od tego, czy klucz jest już skonfigurowany.
"""
provider = (name or default_provider_name()).lower()
if provider not in PROVIDERS:
provider = default_provider_name()
chosen = (model or "").strip() or setting(provider, "MODEL", _DEFAULT_MODEL[provider])
return provider, chosen
def build_provider(name: str | None = None, model: str | None = None) -> LLMProvider:
"""Dostawca modelu. `model` z żądania wygrywa nad konfiguracją — użytkownik
wybiera model w UI, a konfiguracja podaje tylko wartość domyślną."""
name = (name or default_provider_name()).lower()
if name not in PROVIDERS:
raise LLMError(f"Nieznany dostawca LLM: {name!r} (dostępne: {', '.join(PROVIDERS)})")
model = (model or "").strip() or setting(name, "MODEL", _DEFAULT_MODEL[name])
base_url = setting(name, "BASE_URL", _DEFAULT_URL[name])
api_key = setting(name, "API_KEY")
if name in (OPENAI, ANTHROPIC) and not api_key:
raise LLMError(
f"Brak klucza dla dostawcy {name} — ustaw {name.upper()}_API_KEY "
f"(z sekretu). Model lokalny klucza nie wymaga."
)
if name == ANTHROPIC:
return AnthropicProvider(base_url, model, api_key, timeout())
if name == OPENAI:
return ChatCompletionsProvider(OPENAI, base_url, model, api_key, timeout(),
leaves_lan=True)
# lokalny — klucz zwykle zbędny; treść NIE opuszcza sieci
return ChatCompletionsProvider(LOCAL, base_url, model, api_key, timeout(),
leaves_lan=False)
-133
View File
@@ -1,133 +0,0 @@
"""Okna kontekstu modeli i planowanie budżetu tokenów (LOG-30/31).
Po co to istnieje: horoskop MA powstać niezależnie od objętości promptu. Żeby to
zagwarantować, trzeba wiedzieć dwie rzeczy o każdym modelu — ile zmieści na
wejściu (okno kontekstu) i ile maksymalnie wypisze na wyjściu. Bez tego łatwo
wysłać prompt, który wypełnia całe okno i **nie zostawia miejsca na odpowiedź** —
model kończy wtedy na `max_tokens` z pustą albo uciętą treścią.
Zasada naczelna: **zawsze rezerwuj miejsce na odpowiedź.** Budżet promptu liczy
się jako `okno_kontekstu zarezerwowane_wyjście margines`, nigdy odwrotnie.
Wartości są zaszyte jako rozsądne domyślne i nadpisywalne środowiskiem
(`<DOSTAWCA>_CONTEXT_WINDOW`, `<DOSTAWCA>_MAX_OUTPUT`) — modele wychodzą szybciej,
niż aktualizuje się ten plik.
"""
from __future__ import annotations
import os
# model -> (okno kontekstu, maksymalne wyjście) w tokenach
_MODEL_LIMITS: dict[str, tuple[int, int]] = {
# Anthropic
"claude-opus-4-8": (1_000_000, 128_000),
"claude-opus-4-7": (1_000_000, 128_000),
"claude-opus-4-6": (1_000_000, 128_000),
"claude-sonnet-5": (1_000_000, 128_000),
"claude-sonnet-4-6": (1_000_000, 128_000),
"claude-fable-5": (1_000_000, 128_000),
"claude-haiku-4-5": (200_000, 64_000),
# OpenAI
"gpt-4o": (128_000, 16_384),
"gpt-4o-mini": (128_000, 16_384),
"gpt-4.1": (1_000_000, 32_768),
"gpt-4.1-mini": (1_000_000, 32_768),
# lokalne (Ollama/vLLM) — zwykle małe okno, dlatego ostrożna domyślna
"llama3.1": (8_192, 4_096),
"llama3.2": (8_192, 4_096),
"qwen2.5": (32_768, 8_192),
"mistral": (32_768, 8_192),
}
# gdy modelu nie ma w tabeli — zachowawczo, żeby nie obiecywać nieistniejącego okna
_FALLBACK: dict[str, tuple[int, int]] = {
"anthropic": (200_000, 32_000),
"openai": (128_000, 16_384),
"local": (8_192, 4_096),
}
# ile tokenów zostawiamy jako bufor na narzut protokołu i niedokładność liczenia
SAFETY_MARGIN = 2_000
# poniżej tylu tokenów wyjścia nie ma sensu wołać modelu — nie zmieści horoskopu
MIN_OUTPUT = 1_500
# powyżej tylu tokenów promptu ostrzegamy użytkownika (nadal pozwalając wysłać)
WARN_PROMPT_TOKENS = 90_000
def _env_int(provider: str, suffix: str) -> int | None:
raw = os.getenv(f"{provider.upper()}_{suffix}")
if not raw:
return None
try:
value = int(raw)
except ValueError:
return None
return value if value > 0 else None
def limits_for(provider: str, model: str) -> tuple[int, int]:
"""(okno kontekstu, maksymalne wyjście) dla modelu — z nadpisaniem z ENV.
Dopasowanie po prefiksie, bo nazwy modeli lokalnych niosą tag (`llama3.1:8b`).
"""
env_ctx = _env_int(provider, "CONTEXT_WINDOW")
env_out = _env_int(provider, "MAX_OUTPUT")
key = (model or "").strip().lower()
known: tuple[int, int] | None = _MODEL_LIMITS.get(key)
if known is None:
for name, pair in _MODEL_LIMITS.items():
if key.startswith(name):
known = pair
break
if known is None:
known = _FALLBACK.get(provider, _FALLBACK["local"])
return (env_ctx or known[0], env_out or known[1])
def plan(provider: str, model: str, prompt_tokens: int,
want_output: int | None = None) -> dict:
"""Ile tokenów wyjścia zamówić dla promptu tej wielkości.
Zwraca plan z jawną diagnostyką — UI ma z czego zbudować ostrzeżenie, a błąd
ma czym wytłumaczyć, dlaczego się nie udało.
"""
context_window, model_max_output = limits_for(provider, model)
room = context_window - prompt_tokens - SAFETY_MARGIN
target = want_output or model_max_output
max_output = max(0, min(model_max_output, target, room))
warnings: list[str] = []
if prompt_tokens > WARN_PROMPT_TOKENS:
warnings.append(
f"Prompt ma ~{prompt_tokens} tokenów — to dużo. Zapytanie zostanie wysłane, "
f"ale potrwa dłużej i będzie odpowiednio kosztowne."
)
if max_output < MIN_OUTPUT:
warnings.append(
f"Po zmieszczeniu promptu zostaje tylko {max_output} tokenów na odpowiedź "
f"(minimum {MIN_OUTPUT}). Zmniejsz budżet promptu albo wybierz model "
f"z większym oknem kontekstu."
)
return {
"provider": provider,
"model": model,
"context_window": context_window,
"model_max_output": model_max_output,
"prompt_tokens": prompt_tokens,
"max_output": max_output,
"fits": max_output >= MIN_OUTPUT,
"warnings": warnings,
}
def prompt_token_budget(provider: str, model: str, reserve_output: int | None = None) -> int:
"""Ile tokenów promptu wolno wysłać, ZAWSZE zostawiając miejsce na odpowiedź.
To jest podstawa opcji „maksymalny kontekst modelu" w UI.
"""
context_window, model_max_output = limits_for(provider, model)
reserve = reserve_output or model_max_output
return max(0, context_window - reserve - SAFETY_MARGIN)
-306
View File
@@ -1,306 +0,0 @@
"""Implementacje dostawców LLM (LOG-31) — na samym httpx, bez SDK.
Świadomie bez bibliotek `openai` / `anthropic`: lokalny serwer modelu (Ollama,
vLLM, llama.cpp) i OpenAI mówią **tym samym** protokołem `/chat/completions`,
więc jedna implementacja obsługuje oba — różni je tylko adres i klucz. Anthropic
ma własny kształt `/v1/messages`, stąd druga klasa. Mniej zależności, mniej
powierzchni ataku, pełna kontrola nad tym, co wychodzi z sieci.
**Gwarancja niepustej odpowiedzi.** Horoskop ma powstać niezależnie od objętości
promptu, więc `generate()` nie jest pojedynczym strzałem, tylko pętlą:
1. wyślij turę z policzonym limitem wyjścia,
2. jeśli model urwał na limicie — dopisz turę „kontynuuj" i sklej tekst,
3. jeśli tura nie dała ani znaku tekstu — ponów z podpowiedzią,
4. dopiero brak tekstu po wszystkich próbach jest błędem (z diagnostyką).
Kontynuacja jest pewniejsza niż jedno wielkie żądanie: każda tura mieści się
w timeoucie HTTP, a długość odpowiedzi przestaje być ograniczona jedną turą.
**Anthropic i myślenie.** Modele Claude potrafią mieć włączone myślenie, którego
tokeny liczą się do `max_tokens`. Przy ciasnym limicie cała tura potrafi wyjść
jako same bloki `thinking` z pustym tekstem — dokładnie ten objaw, który
zgłoszono. Traktujemy taką turę jak ucięcie i kontynuujemy, zamiast zwracać pustkę.
"""
from __future__ import annotations
import os
import time
import httpx
from app.llm.base import Completion, LLMError, LLMProvider
RETRY_STATUSES = {429, 500, 502, 503, 504}
MAX_ATTEMPTS = 3
# ile razy wolno poprosić model o dokończenie urwanej odpowiedzi
MAX_CONTINUATIONS = 12
# ile tokenów zamawiać na jedną turę — mieści się w timeoucie, a pętla i tak
# dociągnie resztę; zbyt duża wartość ryzykuje zerwanie połączenia w trakcie
TURN_TOKENS_CAP = 16_000
_CONTINUE = (
"Kontynuuj dokładnie od miejsca, w którym przerwałeś — nie powtarzaj tego, "
"co już napisałeś, i nie zaczynaj od nowa. Jeśli skończyłeś całą odpowiedź, "
"napisz wyłącznie: KONIEC"
)
_NUDGE = (
"Nie otrzymałem żadnej treści. Napisz odpowiedź zgodnie z powyższym poleceniem, "
"zaczynając od razu od treści horoskopu."
)
_DONE_MARKER = "KONIEC"
def _post_with_retry(url: str, headers: dict, payload: dict, timeout: float) -> dict:
"""POST z ponawianiem i backoffem — chroni przed chwilowym 429/5xx."""
last: Exception | None = None
for attempt in range(MAX_ATTEMPTS):
try:
with httpx.Client(timeout=timeout) as client:
r = client.post(url, json=payload, headers=headers)
if r.status_code in RETRY_STATUSES and attempt < MAX_ATTEMPTS - 1:
time.sleep(2 ** attempt)
continue
if r.status_code >= 400:
raise LLMError(f"Model odpowiedział błędem {r.status_code}: {r.text[:300]}")
return r.json()
except httpx.TimeoutException as e:
last = e
if attempt < MAX_ATTEMPTS - 1:
time.sleep(2 ** attempt)
continue
raise LLMError(
f"Model nie odpowiedział w czasie {timeout:.0f}s. Zwiększ LLM_TIMEOUT "
f"albo zmniejsz budżet promptu."
) from e
except httpx.HTTPError as e:
last = e
if attempt < MAX_ATTEMPTS - 1:
time.sleep(2 ** attempt)
continue
raise LLMError(f"Nie udało się połączyć z modelem: {e}") from e
raise LLMError(f"Nie udało się wywołać modelu: {last}")
def _merge_usage(total: dict, turn: dict) -> dict:
"""Sumuje zużycie tokenów przez wszystkie tury jednej odpowiedzi."""
for key, value in (turn or {}).items():
if isinstance(value, int):
total[key] = total.get(key, 0) + value
return total
def _join(parts: list[str]) -> str:
return "".join(parts).strip()
def _explain_empty(turns: int, usage: dict, stop: str | None) -> str:
detail = []
if stop:
detail.append(f"powód zakończenia: {stop}")
for key in ("completion_tokens", "output_tokens"):
if usage.get(key) is not None:
detail.append(f"tokeny odpowiedzi: {usage[key]}")
break
suffix = f" ({', '.join(detail)})" if detail else ""
return (
f"Model nie zwrócił żadnej treści po {turns} próbach{suffix}. "
f"Najczęstsza przyczyna: prompt wypełnił okno kontekstu i nie zostało miejsca "
f"na odpowiedź. Zmniejsz budżet promptu albo wybierz model z większym oknem."
)
class _Driver:
"""Wspólna pętla: tura → ewentualna kontynuacja → sklejony tekst.
Podklasy dostarczają tylko `_turn()` — reszta (kontynuacje, ponawianie pustej
tury, sumowanie zużycia) jest identyczna dla obu protokołów.
"""
name: str
model: str
leaves_lan: bool
def _turn(self, messages: list[dict], max_tokens: int):
"""(tekst, czy_ucięta, zużycie, nazwa_modelu, powód_zakończenia)."""
raise NotImplementedError
def generate(self, prompt: str, max_tokens: int) -> Completion:
messages: list[dict] = [{"role": "user", "content": prompt}]
parts: list[str] = []
usage: dict = {}
model_name = self.model
remaining = max(max_tokens, 256)
stop: str | None = None
turns = 0
nudged = False
while turns <= MAX_CONTINUATIONS:
turns += 1
budget = max(256, min(remaining, TURN_TOKENS_CAP))
text, truncated, turn_usage, model_name, stop = self._turn(messages, budget)
_merge_usage(usage, turn_usage)
remaining -= budget
chunk = text.strip()
if chunk:
if chunk.endswith(_DONE_MARKER): # model zgłasza koniec
parts.append(("\n" if parts else "") + chunk[: -len(_DONE_MARKER)].rstrip())
break
parts.append(("\n" if parts else "") + chunk)
if not truncated:
break
elif not truncated:
# pusta i NIE ucięta: jedna próba z podpowiedzią, potem koniec
if nudged or parts:
break
nudged = True
messages = messages + [
{"role": "assistant", "content": ""},
{"role": "user", "content": _NUDGE},
]
continue
# ucięta (także tura złożona z samego myślenia) — poproś o dokończenie
if remaining < 256:
break
messages = [
{"role": "user", "content": prompt},
{"role": "assistant", "content": _join(parts) or ""},
{"role": "user", "content": _CONTINUE},
]
final = _join(parts)
if not final:
raise LLMError(_explain_empty(turns, usage, stop))
usage["turns"] = turns
return Completion(text=final, model=model_name, provider=self.name,
leaves_lan=self.leaves_lan, usage=usage)
def count_tokens(self, prompt: str) -> int:
"""Szacunek tokenów promptu. Dostawcy z własnym licznikiem nadpisują."""
return int(len(prompt) / 3.6)
class ChatCompletionsProvider(_Driver, LLMProvider):
"""Protokół OpenAI `/chat/completions` — lokalny serwer modelu ORAZ OpenAI."""
def __init__(self, name: str, base_url: str, model: str, api_key: str = "",
timeout: float = 120.0, leaves_lan: bool = True) -> None:
self.name = name
self.base_url = base_url.rstrip("/")
self.model = model
self.api_key = api_key
self.timeout = timeout
self.leaves_lan = leaves_lan
def _headers(self) -> dict:
h = {"Content-Type": "application/json"}
if self.api_key:
h["Authorization"] = f"Bearer {self.api_key}"
return h
def _turn(self, messages: list[dict], max_tokens: int):
data = _post_with_retry(
f"{self.base_url}/chat/completions", self._headers(),
{"model": self.model, "max_tokens": max_tokens, "messages": messages},
self.timeout,
)
try:
choice = data["choices"][0]
text = choice["message"].get("content") or ""
except (KeyError, IndexError, TypeError) as e:
raise LLMError(f"Nieoczekiwany kształt odpowiedzi modelu: {str(data)[:300]}") from e
stop = choice.get("finish_reason")
return (text, stop == "length", data.get("usage") or {},
data.get("model", self.model), stop)
def health(self) -> dict:
info = {"provider": self.name, "model": self.model, "leaves_lan": self.leaves_lan}
try:
with httpx.Client(timeout=min(self.timeout, 10.0)) as client:
r = client.get(f"{self.base_url}/models", headers=self._headers())
info["status"] = "ok" if r.status_code < 400 else f"http {r.status_code}"
except httpx.HTTPError as e:
info["status"] = f"down: {e}"
return info
class AnthropicProvider(_Driver, LLMProvider):
"""Protokół Anthropic `/v1/messages`."""
leaves_lan = True
def __init__(self, base_url: str, model: str, api_key: str = "",
timeout: float = 120.0) -> None:
self.name = "anthropic"
self.base_url = base_url.rstrip("/")
self.model = model
self.api_key = api_key
self.timeout = timeout
def _headers(self) -> dict:
return {
"Content-Type": "application/json",
"x-api-key": self.api_key,
"anthropic-version": "2023-06-01",
}
def _thinking(self) -> dict:
"""Konfiguracja myślenia. Domyślnie adaptacyjne — podnosi jakość tekstu.
UWAGA: tokeny myślenia liczą się do `max_tokens`, więc przy ciasnym limicie
cała tura potrafi wyjść jako samo myślenie z pustym tekstem. Pętla
kontynuacji to obsługuje, ale ANTHROPIC_THINKING=off wyłącza myślenie,
gdy zależy nam na przewidywalnym zużyciu tokenów.
"""
mode = os.getenv("ANTHROPIC_THINKING", "adaptive").lower()
if mode in ("off", "disabled", "0", "false"):
return {"thinking": {"type": "disabled"}}
return {
"thinking": {"type": "adaptive"},
"output_config": {"effort": os.getenv("ANTHROPIC_EFFORT", "high")},
}
def _turn(self, messages: list[dict], max_tokens: int):
if not self.api_key:
raise LLMError("Brak ANTHROPIC_API_KEY — dostawca anthropic wymaga klucza.")
payload = {"model": self.model, "max_tokens": max_tokens, "messages": messages}
payload.update(self._thinking())
data = _post_with_retry(f"{self.base_url}/v1/messages", self._headers(),
payload, self.timeout)
try:
blocks = data["content"]
text = "".join(b.get("text", "") for b in blocks if b.get("type") == "text")
except (KeyError, TypeError) as e:
raise LLMError(f"Nieoczekiwany kształt odpowiedzi modelu: {str(data)[:300]}") from e
stop = data.get("stop_reason")
# tura złożona z samego myślenia = budżet poszedł na rozumowanie; traktujemy
# jak ucięcie, żeby pętla poprosiła o treść zamiast zwrócić pustkę
thinking_only = not text.strip() and any(
b.get("type") in ("thinking", "redacted_thinking") for b in blocks
)
return (text, stop == "max_tokens" or thinking_only, data.get("usage") or {},
data.get("model", self.model), stop)
def count_tokens(self, prompt: str) -> int:
"""Dokładny licznik Anthropic — nie szacunek. Od tego zależy, czy po
zmieszczeniu promptu zostanie miejsce na odpowiedź."""
if not self.api_key:
return super().count_tokens(prompt)
try:
data = _post_with_retry(
f"{self.base_url}/v1/messages/count_tokens", self._headers(),
{"model": self.model, "messages": [{"role": "user", "content": prompt}]},
min(self.timeout, 30.0),
)
return int(data.get("input_tokens") or super().count_tokens(prompt))
except LLMError:
return super().count_tokens(prompt)
def health(self) -> dict:
return {
"provider": self.name, "model": self.model, "leaves_lan": True,
"status": "ok (klucz ustawiony)" if self.api_key else "brak ANTHROPIC_API_KEY",
}
+1 -177
View File
@@ -12,17 +12,12 @@ import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from app import link_crypto, security
from app.clients.data_client import DataClient
from app.models import QueryRequest, QueryResponse
from app.service import QueryService
app = FastAPI(title="astrololo · warstwa logiczna")
service = QueryService()
security.install(app, "logiczna") # token międzywarstwowy (LOG-32)
# Szyfrowanie łącza od prezentacji. PO `security.install`, żeby także odmowa
# tokenowa wracała zaszyfrowana — inaczej klient nie umiałby jej odczytać.
link_crypto.install(app, link_crypto.ENV_PRESENTATION_LOGIC, "logiczna")
# --- silnik efemeryd (LOG-24): budowany leniwie, by nie wymagać Skyfielda do startu ---
_engine = None
@@ -44,7 +39,6 @@ class PositionsRequest(BaseModel):
objects: list[str] | None = None
house_system: str = "whole_sign" # whole_sign | equal | porphyry
stations: bool = False # licz stacje (LOG-03; wolniejsze — root-findy)
zodiac: str = "tropical" # LOG-04: tropical | sidereal_{lahiri,fagan_bradley,krishnamurti} | draconic
@app.post("/api/query", response_model=QueryResponse)
@@ -64,10 +58,7 @@ def chart_positions(req: PositionsRequest) -> dict:
engine = get_engine()
moment = ChartMoment(when_utc=req.when_utc, lat=req.lat, lon=req.lon)
try:
chart = build_chart(engine, moment, req.house_system, zodiac=req.zodiac)
except ValueError as e:
raise HTTPException(status_code=422, detail=str(e))
chart = build_chart(engine, moment, req.house_system)
if req.stations:
from app.engine.stations import find_stations
@@ -126,173 +117,6 @@ def chart_report(req: ReportRequest) -> dict:
return {"engine": engine.name, **report}
class PromptRequest(BaseModel):
"""Wejście generatora promptu (LOG-29/30)."""
profile: str = "natal" # natal | period
when_utc: datetime
lat: float = 0.0
lon: float = 0.0
budget: str = "medium" # concise | medium | extensive | huge | max
limit: int = 5000
provider: str | None = None # do wyliczenia budżetu „max" wg okna modelu
model: str | None = None
# tylko dla profilu period:
from_date: str | None = None
to_date: str | None = None
techniques: list[str] | None = None
@app.post("/chart/prompt")
def chart_prompt(req: PromptRequest) -> dict:
"""Gotowy prompt do LLM z naszych wyliczeń (LOG-29) z budżetowaniem (LOG-30).
profile=natal → horoskop urodzeniowy (ekran Interpretacje)
profile=period → horoskop na wybrany okres (ekran Kalendarz)
Nie woła żadnego modelu — zwraca sam prompt i statystyki redukcji, żeby dało się
go obejrzeć i skopiować. Wysyłkę do modelu doda LOG-31.
"""
from app.engine.chart import build_chart
from app.engine.models import ChartMoment
from app.prompt import CHARS_PER_TOKEN, MAX_BUDGET, build_natal_prompt, build_period_prompt
engine = get_engine()
moment = ChartMoment(when_utc=req.when_utc, lat=req.lat, lon=req.lon)
chart = build_chart(engine, moment)
label = req.when_utc.strftime("%Y-%m-%d %H:%M UTC")
data_error = None
# Budżet „maksymalny kontekst modelu": limit znaków liczymy z okna kontekstu
# WYBRANEGO modelu, zawsze po odjęciu miejsca zarezerwowanego na odpowiedź.
budget_chars = None
if req.budget == MAX_BUDGET:
from app.llm.factory import resolve_model
from app.llm.limits import prompt_token_budget
# celowo bez build_provider(): budżet zależy TYLKO od okna kontekstu modelu,
# więc nie może wymagać skonfigurowanego klucza API
provider_name, model_name = resolve_model(req.provider, req.model)
budget_chars = int(prompt_token_budget(provider_name, model_name) * CHARS_PER_TOKEN)
# Warstwa danych dokłada wyłącznie WSKAZANIA. Wyliczenia (horoskop, oś czasu) są
# od niej niezależne — gdy padnie, prompt musi zachować wszystko, co policzyliśmy.
try:
if req.profile == "natal":
from app.significators import build_report
report: dict = {"objects": []}
try:
report = build_report(
chart["positions"], DataClient(),
aspects=chart.get("aspects"), per_object_limit=req.limit,
)
except httpx.HTTPError as e:
data_error = f"Warstwa danych niedostępna: {e}"
out = build_natal_prompt(chart, report, req.budget, label, budget_chars)
elif req.profile == "period":
if not (req.from_date and req.to_date):
raise HTTPException(422, "profile=period wymaga from_date i to_date")
from app.engine import houses as H
from app.engine.timeline import build_timeline
from app.significators import interpret_events
ramc, eps = engine.sidereal(moment)
points = {"Asc": H.compute_asc(ramc, eps, moment.lat), "MC": H.compute_mc(ramc, eps)}
for p in engine.positions(moment):
points[p.name] = p.longitude
events = build_timeline(engine, moment, points, req.from_date, req.to_date,
req.techniques)
try:
interpret_events(events, DataClient())
except httpx.HTTPError as e:
data_error = f"Warstwa danych niedostępna: {e}" # oś czasu zostaje
out = build_period_prompt(chart, events, req.from_date, req.to_date,
req.budget, label, budget_chars)
else:
raise HTTPException(422, f"Nieznany profil: {req.profile!r} (natal | period)")
except ValueError as e: # nieznany budżet
raise HTTPException(422, str(e))
out["engine"] = engine.name
if data_error:
out["data_error"] = data_error
return out
class HoroscopeRequest(PromptRequest):
"""Jak PromptRequest (niesie już provider i model) + limit wyjścia (LOG-31)."""
max_tokens: int | None = None
@app.post("/chart/horoscope")
def chart_horoscope(req: HoroscopeRequest) -> dict:
"""Napisany horoskop (LOG-31): prompt z LOG-29/30 → model → gotowy tekst.
Zwraca TAKŻE użyty prompt — również gdy wywołanie modelu padnie — żeby dało się
go obejrzeć i użyć ręcznie. `leaves_lan` mówi, czy treść baz opuściła naszą sieć
(LOG-32); prezentacja ma na tej podstawie ostrzegać.
"""
from app.llm.base import LLMError
from app.llm.factory import build_provider
from app.llm.limits import plan
out = chart_prompt(req) # ten sam prompt co w podglądzie
try:
provider = build_provider(req.provider, req.model)
# Ile tokenów ma naprawdę ten prompt i ile zostaje na odpowiedź. Anthropic
# liczy dokładnie (własny endpoint), reszta szacuje — od tego zależy, czy
# w oknie kontekstu w ogóle zmieści się miejsce na horoskop.
prompt_tokens = provider.count_tokens(out["prompt"])
budget = plan(provider.name, provider.model, prompt_tokens, req.max_tokens)
out["token_plan"] = budget
if budget["warnings"]:
out["warnings"] = budget["warnings"]
if not budget["fits"]:
out["llm_error"] = " ".join(budget["warnings"])
return out
result = provider.generate(out["prompt"], budget["max_output"])
except LLMError as e:
# prompt zostaje — użytkownik moze go skopiowac i uzyc recznie
out["llm_error"] = str(e)
return out
out.update(
horoscope=result.text,
provider=result.provider,
model=result.model,
leaves_lan=result.leaves_lan,
usage=result.usage,
)
return out
@app.get("/llm/models")
def llm_models() -> dict:
"""Podpowiedzi modeli per dostawca — UI buduje z tego listę wyboru.
To nie jest lista zamknięta: pole modelu jest tekstowe, więc można wpisać
dowolny identyfikator, do którego konto ma dostęp.
"""
from app.llm.catalog import catalog
from app.llm.factory import _DEFAULT_MODEL
return {"providers": catalog(), "defaults": dict(_DEFAULT_MODEL)}
@app.get("/llm/health")
def llm_health(provider: str | None = None) -> dict:
"""Czy model jest osiągalny i skonfigurowany (bez generowania czegokolwiek)."""
from app.llm.base import LLMError
from app.llm.factory import build_provider
try:
return build_provider(provider).health()
except LLMError as e:
return {"status": f"blad konfiguracji: {e}"}
class ProfectionsRequest(BaseModel):
when_utc: datetime # moment urodzenia (UTC)
lat: float = 0.0
-352
View File
@@ -1,352 +0,0 @@
"""Generator promptów do LLM (LOG-29) + budżetowanie rozmiaru (LOG-30).
Składa z naszych wyliczeń gotowe, profesjonalne polecenie po polsku:
- profil **natal** — horoskop urodzeniowy (na bazie /chart/report),
- profil **period** — horoskop na wybrany okres (na bazie /chart/timeline).
Zasada naczelna promptu: model ma pisać WYŁĄCZNIE na podstawie dostarczonych danych
i przy każdej tezie wskazać konkretny sygnifikator, z którego ona wynika. To odróżnia
wynik od ogólnikowej wróżby i pozwala go zweryfikować.
Budżetowanie (LOG-30) — kolejność redukcji:
1. deduplikacja (ten sam sygnifikator ORAZ ten sam opis),
2. grupowanie identycznych opisów z licznikiem wystąpień,
3. sortowanie malejąco wg punktacji siły (LOG-21),
4. obcięcie ogona do budżetu — jednostką obcięcia jest CAŁE wskazanie,
5. skracanie nadmiernie długich opisów (z jawnym oznaczeniem).
Zawsze raportujemy, ile wskazań weszło i ile pominięto — użytkownik ma wiedzieć,
czego brakuje, i móc zwiększyć budżet.
Generowanie jest deterministyczne: ten sam horoskop + ten sam budżet = ten sam prompt.
"""
from __future__ import annotations
import re
# budżety w znakach (całego promptu); dobrane pod wklejanie do ChatGPT/Claude
BUDGETS: dict[str, int] = {
"concise": 4_000,
"medium": 12_000,
"extensive": 30_000,
"huge": 120_000,
# „maksymalny kontekst modelu" — wyliczany dynamicznie z okna kontekstu
# wybranego modelu, ZAWSZE po odjęciu miejsca zarezerwowanego na odpowiedź.
# Wartość poniżej jest tylko zapasem, gdy limity modelu są nieznane.
"max": 400_000,
}
DEFAULT_BUDGET = "medium"
MAX_BUDGET = "max"
MAX_EFFECT_CHARS = 320 # dłuższe opisy skracamy (krok 5 redukcji)
CHARS_PER_TOKEN = 4.0 # zgrubny szacunek tokenów do podglądu w UI
DISCLAIMER = (
"Zakończ krótką notą: treść jest interpretacją astrologiczną i nie stanowi porady "
"medycznej, prawnej ani finansowej."
)
def _norm(text: str) -> str:
return re.sub(r"\s+", " ", (text or "").strip().lower())
def est_tokens(text: str) -> int:
return int(len(text) / CHARS_PER_TOKEN)
def _shorten(text: str, limit: int = MAX_EFFECT_CHARS) -> tuple[str, bool]:
text = (text or "").strip()
if len(text) <= limit:
return text, False
return text[:limit].rstrip() + " […skrócono]", True
# ----------------------------------------------------------------- wskazania
def natal_indications(report: dict) -> list[dict]:
"""Płaska lista wskazań z raportu natalnego: obiekt + faseta + opis + waga."""
out: list[dict] = []
for obj in report.get("objects") or []:
name = obj.get("object")
for facet in obj.get("facets") or []:
score = float(facet.get("score") or 0.0)
for s in facet.get("samples") or []:
out.append({
"context": f"{name}{facet.get('label')}",
"sort_key": (name or "", str(facet.get("label") or "")),
"score": score,
"significator": s.get("expanded") or s.get("significator") or "",
"effect": s.get("effect") or "",
})
return out
def period_indications(events: list[dict]) -> list[dict]:
"""Wskazania z osi czasu — waga rośnie z liczbą trafień w bazie."""
out: list[dict] = []
for ev in events or []:
head = f"{ev.get('technique')} · {ev.get('significator')} · {ev.get('exact')}"
total = float(ev.get("interpretations_count") or 0)
for s in ev.get("interpretations") or []:
out.append({
"context": head,
"sort_key": (str(ev.get("exact") or ""), str(ev.get("technique") or "")),
"score": total,
"significator": s.get("expanded") or s.get("significator") or "",
"effect": s.get("effect") or "",
"date": ev.get("exact"),
"start": ev.get("start"),
"end": ev.get("end"),
"technique": ev.get("technique"),
})
return out
def reduce_indications(items: list[dict], budget_chars: int) -> tuple[list[dict], dict]:
"""Dedup → grupowanie → sortowanie wg wagi → obcięcie do budżetu.
Zwraca (wybrane wskazania, statystyki). Jednostką obcięcia jest całe wskazanie.
"""
# 1+2: dedup i grupowanie identycznych (kontekst, opis) z licznikiem
grouped: dict[tuple[str, str], dict] = {}
order: list[tuple[str, str]] = []
for it in items:
key = (_norm(it["context"]), _norm(it["effect"]))
g = grouped.get(key)
if g is None:
g = {**it, "count": 0, "significators": []}
grouped[key] = g
order.append(key)
g["count"] += 1
if it["significator"] not in g["significators"]:
g["significators"].append(it["significator"])
merged = [grouped[k] for k in order]
deduped = len(items) - len(merged)
# 3: sortowanie malejąco wg wagi; remis rozstrzygany deterministycznie
merged.sort(key=lambda g: (-g["score"], -g["count"], g["sort_key"], g["significator"]))
# 5 (przed obcięciem, bo wpływa na rozmiar): skracanie długich opisów
shortened = 0
for g in merged:
g["effect"], was = _shorten(g["effect"])
shortened += 1 if was else 0
# 4: obcięcie ogona do budżetu — cała pozycja albo nic
chosen: list[dict] = []
used = 0
for g in merged:
cost = len(_render_indication(g)) + 1
if used + cost > budget_chars and chosen:
break
chosen.append(g)
used += cost
omitted = len(merged) - len(chosen)
stats = {
"source_rows": len(items),
"after_grouping": len(merged),
"deduplicated": deduped,
"included": len(chosen),
"omitted": omitted,
"shortened": shortened,
"min_score_included": round(chosen[-1]["score"], 3) if chosen else None,
"max_score_omitted": round(merged[len(chosen)]["score"], 3) if omitted else None,
}
return chosen, stats
def _render_indication(g: dict) -> str:
times = f" (×{g['count']})" if g.get("count", 1) > 1 else ""
sig = g.get("significator") or ""
return f"- {sig}{g.get('effect')}{times}"
def _render_grouped(chosen: list[dict]) -> str:
"""Renderuje wskazania pogrupowane po kontekście, zachowując kolejność wagi."""
blocks: list[str] = []
seen: dict[str, list[dict]] = {}
order: list[str] = []
for g in chosen:
ctx = g["context"]
if ctx not in seen:
seen[ctx] = []
order.append(ctx)
seen[ctx].append(g)
for ctx in order:
rows = seen[ctx]
blocks.append(f"## {ctx} [waga {rows[0]['score']:.2f}]")
blocks.extend(_render_indication(g) for g in rows)
blocks.append("")
return "\n".join(blocks).rstrip()
# ------------------------------------------------------------- sekcje danych
def _chart_section(chart: dict, moment_label: str | None) -> str:
lines: list[str] = ["# DANE HOROSKOPU"]
meta = []
if moment_label:
meta.append(f"moment: {moment_label}")
if chart.get("zodiac"):
z = chart["zodiac"]
if chart.get("ayanamsha") is not None:
z += f" (ayanamsa {chart['ayanamsha']:.4f}°)"
meta.append(f"zodiak: {z}")
if chart.get("house_system"):
meta.append(f"system domów: {chart['house_system']}")
if chart.get("sect"):
meta.append(f"sekta: {'dzienna' if chart['sect'] == 'day' else 'nocna'}")
if meta:
lines.append(" · ".join(meta))
if chart.get("positions"):
lines.append("\n## Pozycje")
for p in chart["positions"]:
house = f"dom {p['house']}" if p.get("house") else ""
lines.append(f"{p['name']:<12} {p.get('in_sign',''):<16} {house:<8} {p.get('direction','')}")
angles = chart.get("angles") or {}
if angles:
lines.append("\n## Osie")
for key in ("Asc", "MC", "Dsc", "IC"):
a = angles.get(key)
if a:
lines.append(f"{a['name']:<5} {a.get('in_sign','')}")
if chart.get("lots"):
lines.append("\n## Lots (punkty arabskie)")
for lot in chart["lots"]:
lines.append(
f"{lot['name']:<10} {lot.get('in_sign',''):<16} dom {lot.get('house','')}"
f" ({lot.get('formula','')})"
)
if chart.get("aspects"):
lines.append("\n## Aspekty (orb; A = aplikacyjny, S = separacyjny)")
for a in chart["aspects"]:
mark = a.get("as") or ""
lines.append(
f"{a['obj1']} {a['aspect']} {a['obj2']} orb {a.get('orb', 0):.2f}° {mark}"
)
return "\n".join(lines)
def _indications_section(chosen: list[dict], stats: dict) -> str:
if not chosen:
return (
"# WSKAZANIA Z BAZ\n"
"(brak trafień w bazach dla tego horoskopu — oprzyj interpretację wyłącznie "
"na danych horoskopu powyżej)"
)
head = [
"# WSKAZANIA Z BAZ INTERPRETACYJNYCH",
"Wskazania dopasowane do tego horoskopu, uporządkowane od najsilniejszych.",
"Zapis: `- sygnifikator → opis (×ile razy wystąpiło w bazach)`.",
]
if stats.get("omitted"):
head.append(
f"UWAGA: pokazano {stats['included']} najsilniejszych wskazań, pominięto "
f"{stats['omitted']} słabszych (limit długości)."
)
return "\n".join(head) + "\n\n" + _render_grouped(chosen)
# ------------------------------------------------------------------ prompty
_NATAL_TASK = """# ZADANIE
Jesteś doświadczonym astrologiem. Napisz profesjonalną interpretację HOROSKOPU
URODZENIOWEGO po polsku, opierając się wyłącznie na danych podanych niżej."""
_PERIOD_TASK = """# ZADANIE
Jesteś doświadczonym astrologiem. Napisz profesjonalną prognozę astrologiczną
NA WYBRANY OKRES po polsku, opierając się wyłącznie na danych podanych niżej."""
_NATAL_OUTPUT = """# JAK MA WYGLĄDAĆ ODPOWIEDŹ
1. Struktura: (a) portret ogólny, (b) temperament i sekta, (c) obszary życia według domów,
(d) napięcia i wyzwania, (e) zasoby i mocne strony, (f) zwięzłe podsumowanie.
2. KAŻDĄ tezę oprzyj na konkretnym wskazaniu i podaj je w nawiasie, np. „(Moon in 12th house)".
Teza bez wskazania jest niedopuszczalna.
3. Nie dodawaj twierdzeń, których nie da się wywieść z powyższych danych. Nie zmyślaj
pozycji, aspektów ani wskazań; nie korzystaj z wiedzy spoza tego promptu.
4. Gdy wskazania są sprzeczne, powiedz to wprost i wskaż obie strony, zamiast wybierać jedną.
5. Wagę wskazania traktuj jako siłę świadectwa — mocniejsze mają pierwszeństwo w syntezie.
6. Ton rzeczowy i profesjonalny: bez wróżbiarstwa, bez straszenia, bez diagnoz medycznych."""
_PERIOD_OUTPUT = """# JAK MA WYGLĄDAĆ ODPOWIEDŹ
1. Uporządkuj prognozę CHRONOLOGICZNIE; przy każdym okresie podaj daty (start / dokładna / koniec).
2. Dla każdej daty napisz, która technika ją wyznacza (profekcja, solariusz, dyrekcja solar-arc,
Firdaria) i co z niej wynika.
3. KAŻDĄ tezę oprzyj na konkretnym wskazaniu i podaj je w nawiasie. Teza bez wskazania jest
niedopuszczalna.
4. Nie zmyślaj dat ani zdarzeń spoza podanych. Nie korzystaj z wiedzy spoza tego promptu.
5. Rozróżniaj okresy o mocnym świadectwie (wysoka waga, kilka technik zbieżnych w czasie)
od słabych — i powiedz wprost, które są które.
6. Na końcu dodaj krótkie zestawienie: najważniejsze okresy w kolejności ważności.
7. Ton rzeczowy i profesjonalny: bez wróżbiarstwa, bez straszenia, bez diagnoz medycznych."""
def _assemble(task: str, chart_sec: str, ind_sec: str, output: str) -> str:
return "\n\n".join([task, chart_sec, ind_sec, output, f"# ZASTRZEŻENIE\n{DISCLAIMER}"])
def _budget_chars(budget: str, budget_chars: int | None = None) -> int:
"""Limit znaków promptu. `budget_chars` nadpisuje tabelę — używane dla opcji
„maksymalny kontekst modelu", gdzie limit zależy od wybranego modelu i musi
być policzony po odjęciu miejsca zarezerwowanego na odpowiedź."""
if budget not in BUDGETS:
raise ValueError(
f"Nieznany budżet: {budget!r} (dostępne: {', '.join(BUDGETS)})"
)
if budget_chars and budget_chars > 0:
return budget_chars
return BUDGETS[budget]
def _finish(prompt: str, budget: str, limit: int, stats: dict, profile: str) -> dict:
stats = {
**stats,
"profile": profile,
"budget": budget,
"limit_chars": limit,
"chars": len(prompt),
"est_tokens": est_tokens(prompt),
}
return {"profile": profile, "prompt": prompt, "stats": stats}
def build_natal_prompt(chart: dict, report: dict, budget: str = DEFAULT_BUDGET,
moment_label: str | None = None,
budget_chars: int | None = None) -> dict:
"""Prompt na horoskop urodzeniowy (ekran „Interpretacje")."""
limit = _budget_chars(budget, budget_chars)
chart_sec = _chart_section(chart, moment_label)
fixed = len(_NATAL_TASK) + len(chart_sec) + len(_NATAL_OUTPUT) + len(DISCLAIMER) + 200
chosen, stats = reduce_indications(natal_indications(report), max(limit - fixed, 500))
prompt = _assemble(_NATAL_TASK, chart_sec, _indications_section(chosen, stats), _NATAL_OUTPUT)
return _finish(prompt, budget, limit, stats, "natal")
def build_period_prompt(chart: dict, events: list[dict], from_date: str, to_date: str,
budget: str = DEFAULT_BUDGET, moment_label: str | None = None,
budget_chars: int | None = None) -> dict:
"""Prompt na horoskop okresowy (ekran „Kalendarz")."""
limit = _budget_chars(budget, budget_chars)
chart_sec = _chart_section(chart, moment_label)
task = f"{_PERIOD_TASK}\nZakres prognozy: **{from_date}{to_date}**."
ev_lines = ["# OŚ CZASU (techniki predykcyjne)",
"Zapis: technika · sygnifikator · start → dokładna → koniec."]
for ev in events or []:
ev_lines.append(
f"- {ev.get('technique')} · {ev.get('significator')} · "
f"{ev.get('start')}{ev.get('exact')}{ev.get('end')}"
)
events_sec = "\n".join(ev_lines)
fixed = len(task) + len(chart_sec) + len(events_sec) + len(_PERIOD_OUTPUT) + len(DISCLAIMER) + 200
chosen, stats = reduce_indications(period_indications(events), max(limit - fixed, 500))
body = events_sec + "\n\n" + _indications_section(chosen, stats)
prompt = _assemble(task, chart_sec, body, _PERIOD_OUTPUT)
stats["events"] = len(events or [])
return _finish(prompt, budget, limit, stats, "period")
-50
View File
@@ -1,50 +0,0 @@
"""Uwierzytelnianie międzywarstwowe (LOG-32).
Warstwa logiczna oddaje treść baz interpretacyjnych, więc samo zalogowanie w
prezentacji nie wystarczy — bez tego kontrolera wystarczyłoby uderzyć w logikę
z pominięciem UI. Gdy ustawiono INTERNAL_TOKEN, każde żądanie (poza /health)
musi go przynieść w nagłówku X-Astrololo-Token.
Bez INTERNAL_TOKEN kontrola jest wyłączona (dev / zgodność wstecz) — wtedy przy
starcie leci ostrzeżenie.
"""
from __future__ import annotations
import logging
import os
import secrets
from fastapi import Request
from fastapi.responses import JSONResponse
log = logging.getLogger("astrololo.security")
HEADER = "X-Astrololo-Token"
PUBLIC_PATHS = frozenset({"/health"})
def token() -> str:
"""Czytany leniwie — konfiguracja może się zmienić bez importu modułu."""
return os.getenv("INTERNAL_TOKEN", "")
def enabled() -> bool:
return bool(token())
def install(app, layer: str) -> None:
if not enabled():
log.warning(
"UWAGA: INTERNAL_TOKEN nie ustawiony — warstwa %s przyjmuje żądania od "
"kogokolwiek, kto ma do niej dostęp sieciowy.", layer,
)
@app.middleware("http")
async def _guard(request: Request, call_next):
if request.url.path in PUBLIC_PATHS or not enabled():
return await call_next(request)
got = request.headers.get(HEADER, "")
if not secrets.compare_digest(got, token()):
return JSONResponse({"detail": "Brak lub błędny token międzywarstwowy."},
status_code=401)
return await call_next(request)
-2
View File
@@ -22,8 +22,6 @@ PLANET_ABBR = {
"Jupiter": "Ju", "Saturn": "Sa", "Uranus": "Ur", "Neptune": "Ne", "Pluto": "Pl",
# punkty wirtualne — tokeny wg SIGNIFICATORS KEY ([NN, [SN, [Lilith)
"North Node": "NN", "South Node": "SN", "Lilith": "Lilith",
# Lots — w bazie Fortuna występuje jako [PF (Part of Fortune)
"Fortune": "PF", "Spirit": "Spirit",
}
SIGN_TO_ABBR = dict(zip(SIGNS, SIGN_ABBR))
-2
View File
@@ -4,5 +4,3 @@ httpx>=0.28
pydantic>=2.10
# Silnik własny (ścieżka A, permisywny): Skyfield (MIT) + dane JPL (public domain)
skyfield>=1.49
# Szyfrowanie łącza między warstwami (PRE-16): AES-256-GCM + HKDF
cryptography>=44.0
+1 -23
View File
@@ -1,5 +1,5 @@
"""Testy aspektów (LOG-06) — czysta matematyka."""
from app.engine.aspects import RIGID_PAIRS, find_aspects, separation
from app.engine.aspects import find_aspects, separation
def test_separation_wraparound():
@@ -55,28 +55,6 @@ def test_separating_when_moving_apart():
assert a["applying"] is False and a["as"] == "S"
def test_rigid_pair_nodes_not_reported():
# SN = NN + 180° z definicji — trywialna opozycja, nie aspekt
pos = [
{"name": "North Node", "decimal": 42.0, "speed": -0.053},
{"name": "South Node", "decimal": 222.0, "speed": -0.053},
{"name": "Sun", "decimal": 42.5, "speed": 0.96},
]
pairs = {frozenset((a["obj1"], a["obj2"])) for a in find_aspects(pos)}
assert frozenset({"North Node", "South Node"}) not in pairs
# aspekty węzłów do innych obiektów zostają nietknięte
assert frozenset({"Sun", "North Node"}) in pairs
assert frozenset({"Sun", "South Node"}) in pairs
def test_no_rigid_pairs_in_real_chart(own_engine, reference_moment):
from app.engine.chart import build_chart
chart = build_chart(own_engine, reference_moment)
found = [a for a in chart["aspects"] if frozenset((a["obj1"], a["obj2"])) in RIGID_PAIRS]
assert not found, f"trywialne aspekty par sztywnych w horoskopie: {found}"
def test_no_as_flag_without_speeds():
pos = [{"name": "Sun", "decimal": 0.0}, {"name": "Moon", "decimal": 2.0}]
assert "as" not in find_aspects(pos)[0]
-91
View File
@@ -1,91 +0,0 @@
"""Okna kontekstu i planowanie budżetu tokenów (LOG-30/31).
Naczelna zasada, której pilnują te testy: **zawsze zostaje miejsce na odpowiedź**.
Prompt nigdy nie może wypełnić całego okna kontekstu, bo wtedy model kończy na
`max_tokens` z pustą albo uciętą treścią — to był zgłoszony błąd.
"""
import pytest
from app.llm import limits
def test_known_models_have_documented_limits():
ctx, out = limits.limits_for("anthropic", "claude-opus-4-8")
assert ctx == 1_000_000 and out == 128_000
ctx, out = limits.limits_for("anthropic", "claude-haiku-4-5")
assert ctx == 200_000 and out == 64_000
def test_local_model_tag_is_matched_by_prefix():
"""Modele lokalne niosą tag (`llama3.1:8b`) — dopasowanie musi to znieść."""
assert limits.limits_for("local", "llama3.1:8b") == limits.limits_for("local", "llama3.1")
def test_unknown_model_falls_back_conservatively():
ctx, out = limits.limits_for("local", "jakis-egzotyczny-model")
assert ctx == 8_192 and out == 4_096
def test_env_overrides_win(monkeypatch):
"""Modele wychodzą szybciej, niż aktualizuje się tabela."""
monkeypatch.setenv("LOCAL_CONTEXT_WINDOW", "131072")
monkeypatch.setenv("LOCAL_MAX_OUTPUT", "8192")
assert limits.limits_for("local", "llama3.1:8b") == (131_072, 8_192)
def test_env_override_ignores_garbage(monkeypatch):
monkeypatch.setenv("LOCAL_CONTEXT_WINDOW", "nie-liczba")
assert limits.limits_for("local", "llama3.1:8b")[0] == 8_192
# ------------------------------------------------- rezerwa miejsca na odpowiedź
def test_prompt_budget_always_reserves_room_for_answer():
ctx, out = limits.limits_for("anthropic", "claude-opus-4-8")
budget = limits.prompt_token_budget("anthropic", "claude-opus-4-8")
assert budget + out + limits.SAFETY_MARGIN <= ctx
assert budget > 0
def test_small_context_model_still_leaves_room():
budget = limits.prompt_token_budget("local", "llama3.1:8b")
ctx, out = limits.limits_for("local", "llama3.1:8b")
assert budget + out + limits.SAFETY_MARGIN <= ctx
def test_plan_shrinks_output_when_prompt_is_huge():
"""Duży prompt nie może dostać pełnego okna wyjścia — musi się zmieścić."""
ctx, model_out = limits.limits_for("local", "llama3.1:8b")
p = limits.plan("local", "llama3.1:8b", prompt_tokens=6_000)
assert p["max_output"] <= ctx - 6_000 - limits.SAFETY_MARGIN
assert p["max_output"] < model_out
def test_plan_reports_not_fitting_instead_of_failing_silently():
p = limits.plan("local", "llama3.1:8b", prompt_tokens=8_000)
assert p["fits"] is False
assert any("odpowied" in w for w in p["warnings"])
def test_plan_warns_above_90k_but_still_fits():
"""Powyżej 90 tys. tokenów ostrzegamy — ale wysłanie MA być nadal możliwe."""
p = limits.plan("anthropic", "claude-opus-4-8", prompt_tokens=120_000)
assert p["fits"] is True, "duży prompt nadal musi dać się wysłać"
assert p["max_output"] >= limits.MIN_OUTPUT
assert any("dużo" in w for w in p["warnings"])
def test_no_warning_below_threshold():
p = limits.plan("anthropic", "claude-opus-4-8", prompt_tokens=10_000)
assert p["warnings"] == [] and p["fits"] is True
def test_requested_output_is_capped_by_model_maximum():
p = limits.plan("anthropic", "claude-haiku-4-5", prompt_tokens=1_000, want_output=999_999)
assert p["max_output"] == 64_000
@pytest.mark.parametrize("tokens", [0, 1_000, 50_000, 200_000, 900_000])
def test_plan_never_returns_negative_output(tokens):
p = limits.plan("anthropic", "claude-opus-4-8", prompt_tokens=tokens)
assert p["max_output"] >= 0
-351
View File
@@ -1,351 +0,0 @@
"""Szyfrowanie łącza między warstwami (PRE-16).
Sedno: przez to łącze płyną surowe wiersze oryginalnych baz interpretacyjnych.
Testy nie sprawdzają więc tylko, czy „coś się zaszyfrowało i odszyfrowało"
sprawdzają, czy tajnego tekstu FAKTYCZNIE NIE MA w bajtach lecących po sieci
oraz czy każda znana droga na skróty (brak szyfrowania, obcy klucz, podmieniony
bajt, przeklejenie na inny endpoint, odtworzenie po czasie) kończy się odmową.
"""
import json
import pathlib
import time
import pytest
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from starlette.testclient import TestClient
from app import link_crypto
from app.link_crypto import ENV_LOGIC_DATA, ENV_PRESENTATION_LOGIC, Link, LinkError
SECRET = "Saturn w VII domu — opis z oryginalnej bazy interpretacyjnej"
KEY_A = "11" * 32 # hex, 32 bajty
KEY_B = "22" * 32
class Payload(BaseModel):
question: str
def _app(link: Link | None) -> FastAPI:
app = FastAPI()
if link is not None:
app.add_middleware(link_crypto.LinkCryptoMiddleware, link=link, layer="testowa")
@app.get("/health")
def health():
return {"status": "ok"}
@app.post("/search")
def search(payload: Payload):
return {"echo": payload.question, "interpretation": SECRET}
@app.get("/catalog")
def catalog():
return {"models": ["a", "b"]}
@app.post("/stream")
def stream():
def lines():
for i in range(4):
yield json.dumps({"step": i, "note": SECRET}).encode() + b"\n"
return StreamingResponse(lines(), media_type="application/x-ndjson")
return app
@pytest.fixture
def link():
return Link(link_crypto.parse_key(KEY_A))
# ------------------------------------------------------------------- klucze
def test_key_accepts_hex_and_base64():
import base64
raw = bytes(range(32))
assert link_crypto.parse_key(raw.hex()) == raw
assert link_crypto.parse_key(base64.b64encode(raw).decode()) == raw
def test_key_of_wrong_length_is_rejected_loudly():
"""Krótki klucz to nie „słabsze szyfrowanie", tylko błąd konfiguracji."""
with pytest.raises(LinkError, match="32"):
link_crypto.parse_key("aabb")
def test_key_that_is_neither_hex_nor_base64_is_rejected():
with pytest.raises(LinkError, match="hex"):
link_crypto.parse_key("!!! to nie jest klucz !!!")
def test_key_from_env_is_lazy(monkeypatch):
monkeypatch.delenv(ENV_PRESENTATION_LOGIC, raising=False)
assert link_crypto.key_from_env(ENV_PRESENTATION_LOGIC) is None
monkeypatch.setenv(ENV_PRESENTATION_LOGIC, KEY_A)
assert link_crypto.key_from_env(ENV_PRESENTATION_LOGIC) == bytes.fromhex(KEY_A)
def test_directions_use_different_subkeys(link):
"""Żądanie i odpowiedź nie dzielą klucza — powtórzenie jednorazówki w jedną
stronę nie osłabia drugiej."""
stamp = link_crypto.stamp_now()
sealed = link.seal(link_crypto.REQUEST, "/search", stamp, 0, b"tajne")
with pytest.raises(LinkError):
link.open(link_crypto.RESPONSE, "/search", stamp, 0, sealed)
# --------------------------------------------------------- podstawowy obieg
def test_round_trip_delivers_plaintext_to_the_app(link):
with TestClient(_app(link)) as client:
got = link_crypto.call_json(client, "POST", "http://testserver/search",
payload={"question": "Saturn"}, link=link)
assert got["echo"] == "Saturn"
assert got["interpretation"] == SECRET
def test_get_without_body_also_works(link):
"""GET nie ma ciała, ale i tak pieczętujemy pustą kopertę — to ona dowodzi,
że pytający ma klucz, i ona wymusza zaszyfrowanie odpowiedzi."""
with TestClient(_app(link)) as client:
got = link_crypto.call_json(client, "GET", "http://testserver/catalog", link=link)
assert got == {"models": ["a", "b"]}
def test_without_key_traffic_stays_plaintext(link):
"""Dev bez sekretów ma działać jak dotąd — inaczej nikt nie odpali projektu lokalnie."""
with TestClient(_app(None)) as client:
got = link_crypto.call_json(client, "POST", "http://testserver/search",
payload={"question": "Saturn"}, link=None)
assert got["interpretation"] == SECRET
def test_health_stays_open_for_kubernetes_probes(link):
"""Sondy k8s klucza nie mają. Gdyby /health wymagał szyfrowania, literówka
w sekrecie kładłaby pody zamiast pokazać błąd w aplikacji."""
with TestClient(_app(link)) as client:
assert client.get("http://testserver/health").json() == {"status": "ok"}
# ------------------------------------------- czy na kablu naprawdę nic nie widać
def _raw_exchange(client, link, payload):
"""Wysyła zapieczętowane żądanie i zwraca SUROWE bajty obu stron."""
stamp = link_crypto.stamp_now()
body = link_crypto.frame_out(
link.seal(link_crypto.REQUEST, "/search", stamp, 0, json.dumps(payload).encode()))
response = client.request(
"POST", "http://testserver/search", content=body,
headers={link_crypto.HEADER_ENC: link_crypto.VERSION,
link_crypto.HEADER_TS: stamp,
"Content-Type": link_crypto.CONTENT_TYPE})
return body, response
def test_request_bytes_do_not_contain_the_question(link):
with TestClient(_app(link)) as client:
body, _ = _raw_exchange(client, link, {"question": "Saturn w VII"})
assert b"Saturn" not in body
assert b"question" not in body
def test_response_bytes_do_not_contain_the_interpretation(link):
"""To jest właściwy powód istnienia całego modułu."""
with TestClient(_app(link)) as client:
_, response = _raw_exchange(client, link, {"question": "Saturn"})
assert response.status_code == 200
assert SECRET.encode() not in response.content
assert b"interpretation" not in response.content
assert response.headers[link_crypto.HEADER_ENC] == link_crypto.VERSION
assert response.headers["content-type"] == link_crypto.CONTENT_TYPE
# ------------------------------------------------------- drogi na skróty i ataki
def test_plaintext_request_is_refused_when_key_is_set(link):
"""Fail-closed: regresja po stronie klienta nie może oznaczać cichego
powrotu do jawnego ruchu."""
with TestClient(_app(link)) as client:
response = client.post("http://testserver/search", json={"question": "Saturn"})
assert response.status_code == 400
assert SECRET.encode() not in response.content
def test_foreign_key_cannot_read_the_link(link):
"""Klucze są osobne dla każdej pary warstw — przejęcie jednego nie otwiera drugiej."""
intruder = Link(link_crypto.parse_key(KEY_B))
with TestClient(_app(link)) as client:
stamp = link_crypto.stamp_now()
body = link_crypto.frame_out(
intruder.seal(link_crypto.REQUEST, "/search", stamp, 0, b'{"question":"x"}'))
response = client.request(
"POST", "http://testserver/search", content=body,
headers={link_crypto.HEADER_ENC: link_crypto.VERSION,
link_crypto.HEADER_TS: stamp,
"Content-Type": link_crypto.CONTENT_TYPE})
assert response.status_code == 400
assert SECRET.encode() not in response.content
def test_single_flipped_bit_is_rejected(link):
"""GCM uwierzytelnia, więc nie ma wariantu „odszyfrowało się, ale zmienione"."""
with TestClient(_app(link)) as client:
stamp = link_crypto.stamp_now()
sealed = bytearray(link.seal(link_crypto.REQUEST, "/search", stamp, 0,
b'{"question":"x"}'))
sealed[-1] ^= 0x01
response = client.request(
"POST", "http://testserver/search",
content=link_crypto.frame_out(bytes(sealed)),
headers={link_crypto.HEADER_ENC: link_crypto.VERSION,
link_crypto.HEADER_TS: stamp,
"Content-Type": link_crypto.CONTENT_TYPE})
assert response.status_code == 400
def test_frame_cannot_be_replayed_against_another_endpoint(link):
"""Ścieżka wchodzi do materiału uwierzytelnianego, więc podsłuchanej koperty
nie da się przekleić tam, gdzie odpowiedź byłaby ciekawsza."""
stamp = link_crypto.stamp_now()
sealed = link.seal(link_crypto.REQUEST, "/catalog", stamp, 0, b"")
with pytest.raises(LinkError):
link.open(link_crypto.REQUEST, "/search", stamp, 0, sealed)
def test_frames_cannot_be_reordered(link):
"""Numer ramki jest uwierzytelniony — przestawienie kolejności w strumieniu
to błąd, a nie po cichu pomieszany horoskop."""
stamp = link_crypto.stamp_now()
second = link.seal(link_crypto.RESPONSE, "/stream", stamp, 1, b"druga")
with pytest.raises(LinkError):
link.open(link_crypto.RESPONSE, "/stream", stamp, 0, second)
def test_stale_frame_is_refused(link, monkeypatch):
"""Bez okna czasowego podsłuchane żądanie dałoby się odtworzyć kiedykolwiek."""
old = f"{time.time() - link_crypto.MAX_SKEW_SECONDS - 60:.3f}"
with TestClient(_app(link)) as client:
body = link_crypto.frame_out(
link.seal(link_crypto.REQUEST, "/search", old, 0, b'{"question":"x"}'))
response = client.request(
"POST", "http://testserver/search", content=body,
headers={link_crypto.HEADER_ENC: link_crypto.VERSION,
link_crypto.HEADER_TS: old,
"Content-Type": link_crypto.CONTENT_TYPE})
assert response.status_code == 400
def test_truncated_stream_is_an_error_not_silent_loss(link):
stamp = link_crypto.stamp_now()
full = link_crypto.frame_out(link.seal(link_crypto.RESPONSE, "/x", stamp, 0, b"abc"))
with pytest.raises(LinkError, match="urwana"):
link.open_all(link_crypto.RESPONSE, "/x", stamp, full[:-2])
def test_absurd_frame_length_does_not_allocate(link):
"""Zadeklarowana długość pochodzi z sieci — nie wolno jej wierzyć na słowo."""
import struct
with pytest.raises(LinkError, match="rozmiar"):
list(link_crypto.frames_in(struct.pack(">I", 2 ** 31) + b"nic"))
# ---------------------------------------------------------- odpowiedź strumieniowa
def test_streaming_response_survives_encryption(link):
"""Okno postępu dostaje kolejne linie na żywo — muszą dojść po kolei
i w komplecie, mimo że każda jedzie w osobnej kopercie."""
with TestClient(_app(link)) as client:
stamp = link_crypto.stamp_now()
body = link_crypto.frame_out(link.seal(link_crypto.REQUEST, "/stream", stamp, 0, b""))
with client.stream("POST", "http://testserver/stream", content=body,
headers={link_crypto.HEADER_ENC: link_crypto.VERSION,
link_crypto.HEADER_TS: stamp,
"Content-Type": link_crypto.CONTENT_TYPE}) as response:
chunks = list(link_crypto.open_response_stream(response, link))
steps = [json.loads(line) for line in b"".join(chunks).splitlines()]
assert [s["step"] for s in steps] == [0, 1, 2, 3]
assert all(s["note"] == SECRET for s in steps)
def test_incremental_unframing_handles_split_frames(link):
"""Ramka potrafi rozjechać się między dwa odczyty z gniazda — składamy ją
w buforze, zamiast zakładać, że każdy kawałek to komplet."""
stamp = link_crypto.stamp_now()
stream = b"".join(link.seal_stream(link_crypto.RESPONSE, "/x", stamp,
[b"raz", b"dwa", b"trzy"]))
buffer = bytearray()
opened, seq = [], 0
for i in range(0, len(stream), 5): # ciachamy w poprzek ramek
buffer += stream[i:i + 5]
for frame in link_crypto.unframe_incremental(buffer):
opened.append(link.open(link_crypto.RESPONSE, "/x", stamp, seq, frame))
seq += 1
assert opened == [b"raz", b"dwa", b"trzy"]
assert not buffer, "bufor musi zostać pusty — inaczej gdzieś zgubiliśmy ramkę"
# ----------------------------------------------- trzy kopie muszą być identyczne
def test_all_three_services_share_the_same_module():
"""Moduł jest skopiowany do trzech niezależnych usług (nie mają wspólnej
biblioteki). Rozjazd między kopiami objawiłby się dopiero na produkcji jako
„nie da się odszyfrować" — więc pilnujemy tego testem."""
root = pathlib.Path(__file__).resolve().parents[3]
copies = {svc: (root / "services" / svc / "app" / "link_crypto.py")
for svc in ("presentation", "logic", "data")}
missing = [svc for svc, path in copies.items() if not path.is_file()]
assert not missing, f"brak modułu w warstwach: {missing}"
contents = {svc: path.read_bytes() for svc, path in copies.items()}
assert len(set(contents.values())) == 1, (
"kopie link_crypto.py rozjechały się między warstwami: "
+ ", ".join(f"{svc}={len(body)}B" for svc, body in contents.items())
)
def test_env_names_are_two_distinct_keys():
"""Wymóg wprost: osobny klucz dla pary prezentacja-logika i logika-dane."""
assert ENV_PRESENTATION_LOGIC != ENV_LOGIC_DATA
# ------------------------------------------------- klient też musi być fail-closed
#
# To wyszło dopiero z podsłuchu prawdziwego gniazda, nie z testów: przy kliencie
# BEZ klucza serwer owszem odmawiał, ale pytanie leciało po drodze otwartym
# tekstem. Odpowiedź była chroniona — zapytanie już nie.
def test_client_without_key_sends_nothing_when_encryption_required(monkeypatch):
monkeypatch.setenv(link_crypto.ENV_REQUIRED, "true")
sent = []
class Tripwire:
def request(self, *args, **kwargs):
sent.append(args)
raise AssertionError("żądanie NIE powinno opuścić procesu")
with pytest.raises(LinkError, match=link_crypto.ENV_REQUIRED):
link_crypto.call(Tripwire(), "POST", "http://logic/search",
payload={"value": "[Sat"}, link=None)
assert not sent, "treść zapytania wyszłaby jawnym tekstem"
def test_plaintext_still_allowed_in_dev(monkeypatch, link):
"""Bez tej flagi lokalne uruchomienie bez sekretów ma dalej działać."""
monkeypatch.delenv(link_crypto.ENV_REQUIRED, raising=False)
with TestClient(_app(None)) as client:
got = link_crypto.call_json(client, "POST", "http://testserver/search",
payload={"question": "Saturn"}, link=None)
assert got["interpretation"] == SECRET
def test_service_refuses_to_start_without_key_when_required(monkeypatch):
"""Pod w CrashLoop widać od razu; usługę, która wstała i nie szyfruje — nie."""
monkeypatch.setenv(link_crypto.ENV_REQUIRED, "true")
monkeypatch.delenv(ENV_LOGIC_DATA, raising=False)
with pytest.raises(LinkError, match=ENV_LOGIC_DATA):
link_crypto.install(FastAPI(), ENV_LOGIC_DATA, "danych")
-407
View File
@@ -1,407 +0,0 @@
"""Dostawcy LLM (LOG-31) — bez wołania jakiegokolwiek prawdziwego modelu.
Transport podstawiamy przez httpx.MockTransport, więc testy są szybkie,
deterministyczne i nic nie wychodzi na zewnątrz.
"""
import json
import httpx
import pytest
from app.llm import factory
from app.llm.base import Completion, LLMError
from app.llm.providers import AnthropicProvider, ChatCompletionsProvider
def _mock_client(handler):
"""Podmienia httpx.Client na wersję z transportem testowym."""
class _C(httpx.Client):
def __init__(self, *a, **kw):
kw["transport"] = httpx.MockTransport(handler)
super().__init__(*a, **kw)
return _C
@pytest.fixture
def chat_ok(monkeypatch):
def handler(request):
assert request.url.path.endswith("/chat/completions")
return httpx.Response(200, json={
"model": "test-model",
"choices": [{"message": {"content": "Horoskop testowy."}}],
"usage": {"prompt_tokens": 100, "completion_tokens": 50},
})
monkeypatch.setattr(httpx, "Client", _mock_client(handler))
# ------------------------------------------------------- protokół chat/completions
def test_local_provider_generates(chat_ok):
p = ChatCompletionsProvider("local", "http://localhost:11434/v1", "m", leaves_lan=False)
out = p.generate("prompt", 500)
assert isinstance(out, Completion)
assert out.text == "Horoskop testowy."
assert out.usage["completion_tokens"] == 50
def test_local_provider_does_not_leave_lan(chat_ok):
p = ChatCompletionsProvider("local", "http://localhost:11434/v1", "m", leaves_lan=False)
assert p.generate("prompt", 100).leaves_lan is False
def test_cloud_provider_marks_leaving_lan(chat_ok):
p = ChatCompletionsProvider("openai", "https://api.openai.com/v1", "m", "klucz",
leaves_lan=True)
assert p.generate("prompt", 100).leaves_lan is True
def test_api_key_sent_only_when_set(monkeypatch):
seen = {}
def handler(request):
seen["auth"] = request.headers.get("authorization")
return httpx.Response(200, json={"choices": [{"message": {"content": "x"}}]})
monkeypatch.setattr(httpx, "Client", _mock_client(handler))
ChatCompletionsProvider("local", "http://x/v1", "m").generate("p", 10)
assert seen["auth"] is None
ChatCompletionsProvider("openai", "http://x/v1", "m", "tajny").generate("p", 10)
assert seen["auth"] == "Bearer tajny"
def test_http_error_becomes_readable_message(monkeypatch):
monkeypatch.setattr(httpx, "Client",
_mock_client(lambda r: httpx.Response(400, text="zly model")))
with pytest.raises(LLMError, match="400"):
ChatCompletionsProvider("local", "http://x/v1", "m").generate("p", 10)
def test_malformed_response_reported(monkeypatch):
monkeypatch.setattr(httpx, "Client",
_mock_client(lambda r: httpx.Response(200, json={"nonsens": 1})))
with pytest.raises(LLMError, match="kształt"):
ChatCompletionsProvider("local", "http://x/v1", "m").generate("p", 10)
def test_retries_then_succeeds(monkeypatch):
calls = {"n": 0}
def handler(request):
calls["n"] += 1
if calls["n"] < 3:
return httpx.Response(429, text="za duzo")
return httpx.Response(200, json={"choices": [{"message": {"content": "ok"}}]})
monkeypatch.setattr(httpx, "Client", _mock_client(handler))
monkeypatch.setattr("app.llm.providers.time.sleep", lambda s: None) # bez czekania
assert ChatCompletionsProvider("local", "http://x/v1", "m").generate("p", 10).text == "ok"
assert calls["n"] == 3
# ------------------------------------------------------------------- Anthropic
def test_anthropic_generates(monkeypatch):
def handler(request):
assert request.url.path.endswith("/v1/messages")
assert request.headers.get("x-api-key") == "klucz"
assert request.headers.get("anthropic-version")
return httpx.Response(200, json={
"model": "claude-x",
"content": [{"type": "text", "text": "Prognoza."}],
"usage": {"input_tokens": 10, "output_tokens": 5},
})
monkeypatch.setattr(httpx, "Client", _mock_client(handler))
out = AnthropicProvider("https://api.anthropic.com", "claude-x", "klucz").generate("p", 100)
assert out.text == "Prognoza." and out.leaves_lan is True
def test_anthropic_requires_key():
with pytest.raises(LLMError, match="ANTHROPIC_API_KEY"):
AnthropicProvider("https://api.anthropic.com", "m", "").generate("p", 10)
# --------------------------------------------------------------------- fabryka
def test_default_provider_is_local(monkeypatch):
monkeypatch.delenv("LLM_PROVIDER", raising=False)
monkeypatch.delenv("LLM_MODEL", raising=False)
monkeypatch.delenv("LLM_BASE_URL", raising=False)
p = factory.build_provider()
assert p.name == "local"
# domyślnie NIC nie opuszcza sieci — prompt niesie opisy z baz (LOG-32)
assert p.leaves_lan is False
def test_openai_requires_key(monkeypatch):
monkeypatch.delenv("LLM_API_KEY", raising=False)
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
# komunikat wskazuje ZMIENNĄ DO USTAWIENIA dla tego dostawcy, nie ogólne LLM_API_KEY
with pytest.raises(LLMError, match="OPENAI_API_KEY"):
factory.build_provider("openai")
def test_unknown_provider_rejected():
with pytest.raises(LLMError, match="Nieznany dostawca"):
factory.build_provider("bzdura")
def test_env_overrides_model_and_url(monkeypatch):
monkeypatch.setenv("LLM_MODEL", "moj-model")
monkeypatch.setenv("LLM_BASE_URL", "http://serwer:8000/v1")
p = factory.build_provider("local")
assert p.model == "moj-model" and p.base_url == "http://serwer:8000/v1"
# ---------------------------------------- konfiguracja per dostawca (regresja LOG-31)
# UI pozwala przelaczac dostawce przy kazdym zadaniu, wiec ustawienia JEDNEGO nie moga
# przeciekac na pozostalych. Wczesniej wspolne LLM_BASE_URL/LLM_MODEL kierowaly zadania
# do OpenAI na adres lokalnej Ollamy i prosily Anthropic o model llama.
def _clear(monkeypatch):
for v in ("LLM_PROVIDER", "LLM_MODEL", "LLM_BASE_URL", "LLM_API_KEY",
"LOCAL_MODEL", "LOCAL_BASE_URL", "LOCAL_API_KEY",
"OPENAI_MODEL", "OPENAI_BASE_URL", "OPENAI_API_KEY",
"ANTHROPIC_MODEL", "ANTHROPIC_BASE_URL", "ANTHROPIC_API_KEY"):
monkeypatch.delenv(v, raising=False)
def test_local_config_does_not_leak_to_cloud(monkeypatch):
"""Sedno bledu: skonfigurowany model lokalny przejmowal zadania do chmury."""
_clear(monkeypatch)
monkeypatch.setenv("LLM_PROVIDER", "local")
monkeypatch.setenv("LLM_BASE_URL", "http://ollama:11434/v1") # konfiguracja lokalnego
monkeypatch.setenv("LLM_MODEL", "llama3.1:8b")
monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
local = factory.build_provider("local")
assert local.base_url == "http://ollama:11434/v1" and local.model == "llama3.1:8b"
openai = factory.build_provider("openai")
assert openai.base_url == "https://api.openai.com/v1", "zadanie do OpenAI poszloby do Ollamy"
assert openai.model == "gpt-4o-mini", "OpenAI dostalby nazwe modelu llama"
def test_provider_specific_settings_win(monkeypatch):
_clear(monkeypatch)
monkeypatch.setenv("LLM_PROVIDER", "local")
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant")
monkeypatch.setenv("ANTHROPIC_MODEL", "claude-opus-4-8")
p = factory.build_provider("anthropic")
assert p.model == "claude-opus-4-8" and p.api_key == "sk-ant"
def test_generic_vars_apply_only_to_default_provider(monkeypatch):
"""Zgodnosc wstecz: wspolne LLM_* konfiguruja dostawce domyslnego i tylko jego."""
_clear(monkeypatch)
monkeypatch.setenv("LLM_PROVIDER", "openai")
monkeypatch.setenv("LLM_API_KEY", "sk-generic")
monkeypatch.setenv("LLM_MODEL", "gpt-4o")
assert factory.build_provider("openai").model == "gpt-4o"
assert factory.build_provider("local").model == "llama3.1:8b" # nie dziedziczy
def test_cloud_without_key_is_rejected_clearly(monkeypatch):
_clear(monkeypatch)
monkeypatch.setenv("LLM_PROVIDER", "local")
for name in ("openai", "anthropic"):
with pytest.raises(LLMError, match=f"{name.upper()}_API_KEY"):
factory.build_provider(name)
def test_local_needs_no_key(monkeypatch):
_clear(monkeypatch)
assert factory.build_provider("local").api_key == ""
# ------------------------------------------- pusta odpowiedz modelu (cicha awaria)
# Regresja: model potrafi oddac pusta tresc (prompt zjadl caly kontekst ->
# finish_reason=length, completion_tokens=0). Wczesniej generate() zwracalo pusty
# tekst BEZ bledu, widok nic nie renderowal i uzytkownik dostawal pusta strone
# bez zadnego wyjasnienia. Pusta odpowiedz MUSI byc bledem.
def test_empty_completion_raises_instead_of_silent_blank(monkeypatch):
monkeypatch.setattr(httpx, "Client", _mock_client(lambda r: httpx.Response(200, json={
"model": "llama3.1:8b",
"choices": [{"message": {"content": ""}, "finish_reason": "length"}],
"usage": {"prompt_tokens": 8000, "completion_tokens": 0},
})))
with pytest.raises(LLMError, match="nie zwrócił żadnej treści"):
ChatCompletionsProvider("local", "http://x/v1", "m").generate("p", 2000)
def test_empty_completion_explains_context_window(monkeypatch):
"""Komunikat ma prowadzic do przyczyny, a nie tylko stwierdzac fakt."""
monkeypatch.setattr(httpx, "Client", _mock_client(lambda r: httpx.Response(200, json={
"choices": [{"message": {"content": " "}, "finish_reason": "length"}],
"usage": {"prompt_tokens": 8000, "completion_tokens": 0},
})))
with pytest.raises(LLMError) as ei:
ChatCompletionsProvider("local", "http://x/v1", "m").generate("p", 2000)
msg = str(ei.value)
assert "kontekstu" in msg and "budżet" in msg
assert "powód zakończenia: length" in msg # diagnostyka w tresci bledu
def test_whitespace_only_is_treated_as_empty(monkeypatch):
monkeypatch.setattr(httpx, "Client", _mock_client(lambda r: httpx.Response(200, json={
"choices": [{"message": {"content": "\n\n \t "}}],
})))
with pytest.raises(LLMError, match="nie zwrócił żadnej treści"):
ChatCompletionsProvider("local", "http://x/v1", "m").generate("p", 100)
def test_anthropic_empty_completion_raises(monkeypatch):
monkeypatch.setattr(httpx, "Client", _mock_client(lambda r: httpx.Response(200, json={
"model": "claude-x", "content": [], "stop_reason": "max_tokens",
"usage": {"input_tokens": 9000, "output_tokens": 0},
})))
with pytest.raises(LLMError, match="nie zwrócił żadnej treści"):
AnthropicProvider("https://api.anthropic.com", "claude-x", "klucz").generate("p", 100)
def test_normal_response_still_passes(monkeypatch):
"""Straznik nie moze psuc poprawnej odpowiedzi."""
monkeypatch.setattr(httpx, "Client", _mock_client(lambda r: httpx.Response(200, json={
"choices": [{"message": {"content": "Horoskop."}, "finish_reason": "stop"}],
"usage": {"prompt_tokens": 100, "completion_tokens": 20},
})))
assert ChatCompletionsProvider("local", "http://x/v1", "m").generate("p", 100).text == "Horoskop."
# ------------------------------------- kontynuacja: horoskop MA powstac zawsze
# Sedno wymagania: niezaleznie od objetosci promptu i limitu wyjscia, pelna tresc
# ma wrocic do uzytkownika. Model urwany na max_tokens jest proszony o dokonczenie
# w ramach tej samej rozmowy, a kawalki sa sklejane.
def _scripted(responses):
"""Transport oddajacy kolejne odpowiedzi z listy (po jednej na ture)."""
seq = list(responses)
seen = []
def handler(request):
seen.append(request)
return httpx.Response(200, json=seq.pop(0) if seq else seq_last)
seq_last = responses[-1]
return handler, seen
def _chat(text, finish):
return {"choices": [{"message": {"content": text}, "finish_reason": finish}],
"usage": {"completion_tokens": 10}}
def test_truncated_answer_is_continued_and_joined(monkeypatch):
handler, seen = _scripted([
_chat("Czesc pierwsza.", "length"),
_chat("Czesc druga. KONIEC", "stop"),
])
monkeypatch.setattr(httpx, "Client", _mock_client(handler))
out = ChatCompletionsProvider("local", "http://x/v1", "m").generate("prompt", 40000)
assert "Czesc pierwsza." in out.text and "Czesc druga." in out.text
assert "KONIEC" not in out.text # znacznik nie trafia do horoskopu
assert out.usage["turns"] == 2
def test_continuation_asks_in_same_conversation(monkeypatch):
"""Kontynuacja musi isc jako kolejna tura rozmowy, a ostatnia wiadomosc MUSI
byc od uzytkownika — Claude odrzuca prefill w turze asystenta (400)."""
handler, seen = _scripted([_chat("Poczatek", "length"), _chat("Reszta", "stop")])
monkeypatch.setattr(httpx, "Client", _mock_client(handler))
ChatCompletionsProvider("local", "http://x/v1", "m").generate("prompt", 40000)
msgs = json.loads(seen[1].content)["messages"]
assert msgs[-1]["role"] == "user", "ostatnia wiadomosc nie moze byc prefillem asystenta"
assert msgs[1]["role"] == "assistant" and "Poczatek" in msgs[1]["content"]
def test_anthropic_thinking_only_turn_is_continued(monkeypatch):
"""DOKLADNIE zgloszony objaw: cala tura poszla na myslenie, tekst pusty.
Wczesniej konczylo sie to pusta strona; teraz pytamy o tresc dalej."""
seq = [
{"content": [{"type": "thinking", "thinking": ""}], "stop_reason": "max_tokens",
"usage": {"output_tokens": 2000}},
{"content": [{"type": "text", "text": "Horoskop urodzeniowy..."}],
"stop_reason": "end_turn", "usage": {"output_tokens": 500}},
]
def handler(request):
return httpx.Response(200, json=seq.pop(0) if seq else seq[-1])
monkeypatch.setattr(httpx, "Client", _mock_client(handler))
out = AnthropicProvider("https://api.anthropic.com", "claude-opus-4-8", "k").generate("p", 40000)
assert out.text == "Horoskop urodzeniowy..."
assert out.usage["turns"] == 2
def test_anthropic_sends_thinking_config(monkeypatch):
"""Bez jawnego `thinking` Sonnet 5 wlacza myslenie sam — konfigurujemy to wprost."""
seen = []
def handler(request):
seen.append(json.loads(request.content))
return httpx.Response(200, json={"content": [{"type": "text", "text": "ok"}],
"stop_reason": "end_turn"})
monkeypatch.setattr(httpx, "Client", _mock_client(handler))
monkeypatch.delenv("ANTHROPIC_THINKING", raising=False)
AnthropicProvider("https://api.anthropic.com", "claude-opus-4-8", "k").generate("p", 5000)
assert seen[0]["thinking"] == {"type": "adaptive"}
seen.clear()
monkeypatch.setenv("ANTHROPIC_THINKING", "off")
AnthropicProvider("https://api.anthropic.com", "claude-opus-4-8", "k").generate("p", 5000)
assert seen[0]["thinking"] == {"type": "disabled"}
def test_complete_answer_does_not_loop(monkeypatch):
"""Straznik nie moze mnozyc zapytan, gdy model skonczyl normalnie."""
calls = {"n": 0}
def handler(request):
calls["n"] += 1
return httpx.Response(200, json=_chat("Gotowe.", "stop"))
monkeypatch.setattr(httpx, "Client", _mock_client(handler))
out = ChatCompletionsProvider("local", "http://x/v1", "m").generate("p", 40000)
assert out.text == "Gotowe." and calls["n"] == 1
# ------------------------------------------- wybor modelu przez uzytkownika (UI)
def test_model_from_request_wins_over_config(monkeypatch):
_clear(monkeypatch)
monkeypatch.setenv("ANTHROPIC_MODEL", "claude-opus-4-8")
monkeypatch.setenv("ANTHROPIC_API_KEY", "k")
p = factory.build_provider("anthropic", "claude-fable-5")
assert p.model == "claude-fable-5", "wybor z UI musi wygrac nad konfiguracja"
def test_blank_model_falls_back_to_configured_default(monkeypatch):
_clear(monkeypatch)
monkeypatch.setenv("ANTHROPIC_MODEL", "claude-sonnet-5")
monkeypatch.setenv("ANTHROPIC_API_KEY", "k")
assert factory.build_provider("anthropic", " ").model == "claude-sonnet-5"
def test_resolve_model_needs_no_api_key(monkeypatch):
"""Budzet promptu zalezy od okna kontekstu modelu — nie moze wymagac klucza.
Wczesniej liczenie budzetu szlo przez build_provider(), ktory bez klucza
rzuca bledem, wiec „maksymalny kontekst" cicho spadal do wartosci zapasowej.
"""
_clear(monkeypatch)
provider, model = factory.resolve_model("anthropic", "claude-haiku-4-5")
assert (provider, model) == ("anthropic", "claude-haiku-4-5")
with pytest.raises(LLMError): # samo zbudowanie nadal wymaga klucza
factory.build_provider("anthropic", "claude-haiku-4-5")
def test_max_budget_differs_between_models(monkeypatch):
"""Sedno funkcji: wieksze okno = wiekszy budzet promptu."""
from app.llm.limits import prompt_token_budget
_clear(monkeypatch)
opus = prompt_token_budget(*factory.resolve_model("anthropic", "claude-opus-4-8"))
haiku = prompt_token_budget(*factory.resolve_model("anthropic", "claude-haiku-4-5"))
local = prompt_token_budget(*factory.resolve_model("local", "llama3.1:8b"))
assert opus > haiku > local > 0
-71
View File
@@ -1,71 +0,0 @@
"""Lots / punkty arabskie (LOG-08).
Wyrocznie z notes3 (astro-seek, horoskop referencyjny 30.04.1984, urodzenie DZIENNE):
- Fortuna wprost w tabeli obiektów: Cancer 12°35'24" = 102.5900°
- Ducha (Spirit) potwierdza jego antyscja: Taurus 28°14' -> Spirit = 180 58.2333
"""
import pytest
from app.engine.formats import sign_index
from app.engine.lots import LOT_DEFS, compute_lots
# natalne długości horoskopu referencyjnego
NATAL = {"Asc": 112.1786, "Sun": 40.1392, "Moon": 30.5514, "Mercury": 27.3839,
"Venus": 27.6769, "Mars": 234.5411, "Jupiter": 282.9617, "Saturn": 223.3108}
FORTUNE_REF = 102.5900 # Cancer 12°35'24"
SPIRIT_REF = 180.0 - 58.2333 # z antyscji Taurus 28°14'
def _by_name(lots):
return {lot["name"]: lot["longitude"] for lot in lots}
def _arcmin(a, b):
return abs(((a - b + 180.0) % 360.0) - 180.0) * 60.0
def test_fortune_matches_astroseek():
lots = _by_name(compute_lots(NATAL, is_day=True))
assert _arcmin(lots["Fortune"], FORTUNE_REF) < 2.0
assert sign_index(lots["Fortune"]) == 3 # Cancer
def test_spirit_matches_antiscia_reference():
lots = _by_name(compute_lots(NATAL, is_day=True))
assert _arcmin(lots["Spirit"], SPIRIT_REF) < 2.0
def test_fortune_and_spirit_reverse_at_night():
day = _by_name(compute_lots(NATAL, is_day=True))
night = _by_name(compute_lots(NATAL, is_day=False))
# nocą formuła się odwraca: Fortuna nocna = Duch dzienny (i odwrotnie)
assert _arcmin(night["Fortune"], day["Spirit"]) < 0.01
assert _arcmin(night["Spirit"], day["Fortune"]) < 0.01
def test_all_seven_hermetic_lots_present():
lots = compute_lots(NATAL, is_day=True)
assert [lot["name"] for lot in lots] == [d[0] for d in LOT_DEFS]
assert len(lots) == 7
def test_derived_lots_use_fortune_and_spirit():
lots = _by_name(compute_lots(NATAL, is_day=True))
# Necessity = Asc + Fortune Mercury
expected = (NATAL["Asc"] + lots["Fortune"] - NATAL["Mercury"]) % 360.0
assert _arcmin(lots["Necessity"], expected) < 0.01
# Eros = Asc + Venus Spirit
expected_eros = (NATAL["Asc"] + NATAL["Venus"] - lots["Spirit"]) % 360.0
assert _arcmin(lots["Eros"], expected_eros) < 0.01
def test_by_sign_method_lands_on_sign_start():
lots = compute_lots(NATAL, is_day=True, method="sign")
for lot in lots:
assert lot["longitude"] % 30.0 == 0.0 # 0° wyliczonego znaku
def test_unknown_method_rejected():
with pytest.raises(ValueError):
compute_lots(NATAL, is_day=True, method="bzdura")
-218
View File
@@ -1,218 +0,0 @@
"""Generator promptów (LOG-29) i budżetowanie (LOG-30).
Testy nie wołają żadnego modelu — sprawdzają skład promptu i niezmienniki redukcji.
"""
import pytest
from app.prompt import (
BUDGETS,
build_natal_prompt,
build_period_prompt,
natal_indications,
period_indications,
reduce_indications,
)
CHART = {
"zodiac": "tropical",
"house_system": "whole_sign",
"sect": "day",
"positions": [
{"name": "Sun", "sign": "Taurus", "in_sign": "Tau 10°12'37\"", "house": 11, "direction": "D"},
{"name": "Moon", "sign": "Aries", "in_sign": "Ari 7°48'40\"", "house": 10, "direction": "D"},
],
"angles": {"Asc": {"name": "Asc", "in_sign": "Can 16°00'54\""},
"MC": {"name": "MC", "in_sign": "Pis 25°50'50\""}},
"lots": [{"name": "Fortune", "in_sign": "Can 7°15'14\"", "house": 1,
"formula": "Asc + Moon Sun"}],
"aspects": [{"obj1": "Sun", "obj2": "Moon", "aspect": "conjunction", "orb": 2.34, "as": "A"}],
}
def _report(n_samples=3, score=2.0):
return {"objects": [{
"object": "Sun", "sign": "Taurus", "house": 11,
"facets": [{
"type": "sign", "label": "w znaku Taurus", "score": score,
"samples": [{"significator": f"[Su in [Tau {i}", "expanded": f"Sun in Taurus {i}",
"effect": f"efekt numer {i}"} for i in range(n_samples)],
}],
}]}
# ---------------------------------------------------------------- budżetowanie
def test_budget_limits_prompt_size():
# opisy MUSZĄ być różne — identyczne zlałoby grupowanie w jedną pozycję
big = {"objects": [{
"object": "Sun", "sign": "Taurus", "house": 11,
"facets": [{"type": "sign", "label": "w znaku Taurus", "score": 1.0,
"samples": [{"significator": f"[Su {i}", "expanded": f"Sun {i}",
"effect": f"opis {i} " + "x" * 200} for i in range(500)]}],
}]}
out = build_natal_prompt(CHART, big, budget="concise")
assert out["stats"]["chars"] <= BUDGETS["concise"] * 1.1 # z marginesem na sekcje stałe
assert out["stats"]["omitted"] > 0
def test_bigger_budget_includes_more():
big = _report(n_samples=300, score=1.0)
small = build_natal_prompt(CHART, big, budget="concise")["stats"]
large = build_natal_prompt(CHART, big, budget="extensive")["stats"]
assert large["included"] > small["included"]
assert large["omitted"] < small["omitted"]
def test_grouping_merges_identical_effects():
items = [
{"context": "Sun — w znaku", "sort_key": ("Sun", "a"), "score": 1.0,
"significator": "Sun in Taurus", "effect": "ten sam opis"},
{"context": "Sun — w znaku", "sort_key": ("Sun", "a"), "score": 1.0,
"significator": "Sun in Taurus (inny zapis)", "effect": "Ten Sam Opis "},
]
chosen, stats = reduce_indications(items, 10_000)
assert len(chosen) == 1 # zlane w jedno
assert chosen[0]["count"] == 2 # z licznikiem wystąpień
assert stats["after_grouping"] == 1
assert stats["deduplicated"] == 1
def test_sorted_by_score_desc():
items = [
{"context": f"c{i}", "sort_key": (f"c{i}", ""), "score": float(i),
"significator": f"s{i}", "effect": f"e{i}"} for i in range(5)
]
chosen, _ = reduce_indications(items, 10_000)
scores = [c["score"] for c in chosen]
assert scores == sorted(scores, reverse=True)
def test_weakest_are_dropped_first():
items = [
{"context": f"c{i}", "sort_key": (f"c{i}", ""), "score": float(i),
"significator": f"s{i}", "effect": "opis " * 20} for i in range(20)
]
chosen, stats = reduce_indications(items, 400)
assert stats["omitted"] > 0
# to, co weszło, ma wagę nie niższą niż to, co odpadło
assert stats["min_score_included"] >= stats["max_score_omitted"]
def test_long_effects_are_shortened():
items = [{"context": "c", "sort_key": ("c", ""), "score": 1.0,
"significator": "s", "effect": "x" * 1000}]
chosen, stats = reduce_indications(items, 10_000)
assert stats["shortened"] == 1
assert "skrócono" in chosen[0]["effect"]
def test_never_truncates_mid_indication():
items = [{"context": "c", "sort_key": ("c", ""), "score": 1.0,
"significator": "s", "effect": "opis " * 50} for _ in range(10)]
chosen, _ = reduce_indications(items, 200)
assert chosen # zawsze co najmniej jedno całe
for c in chosen:
assert c["effect"].endswith(("opis", "[…skrócono]")) # nie urwane w pół słowa
def test_unknown_budget_rejected():
with pytest.raises(ValueError):
build_natal_prompt(CHART, _report(), budget="gigantyczny")
# ----------------------------------------------------------------- skład promptu
def test_natal_prompt_contains_required_sections():
p = build_natal_prompt(CHART, _report(), moment_label="1984-04-30 09:20 UTC")["prompt"]
for section in ["# ZADANIE", "# DANE HOROSKOPU", "# WSKAZANIA Z BAZ",
"# JAK MA WYGLĄDAĆ ODPOWIEDŹ", "# ZASTRZEŻENIE"]:
assert section in p
assert "HOROSKOPU\nURODZENIOWEGO" in p or "URODZENIOWEGO" in p
def test_natal_prompt_carries_chart_data():
p = build_natal_prompt(CHART, _report(), moment_label="1984-04-30 09:20 UTC")["prompt"]
assert "Tau 10°12'37\"" in p and "dom 11" in p # pozycja + dom
assert "Can 16°00'54\"" in p # Asc
assert "Fortune" in p # Lots
assert "Sun conjunction Moon" in p or "conjunction" in p
assert "sekta: dzienna" in p
assert "1984-04-30 09:20 UTC" in p
def test_prompt_demands_citing_significators():
p = build_natal_prompt(CHART, _report())["prompt"]
assert "Teza bez wskazania jest niedopuszczalna" in p
assert "Nie zmyślaj" in p
def test_prompt_has_disclaimer():
p = build_natal_prompt(CHART, _report())["prompt"]
assert "nie stanowi porady" in p
def test_prompt_is_deterministic():
a = build_natal_prompt(CHART, _report(n_samples=50), budget="medium")["prompt"]
b = build_natal_prompt(CHART, _report(n_samples=50), budget="medium")["prompt"]
assert a == b
def test_omission_is_reported_in_prompt():
out = build_natal_prompt(CHART, _report(n_samples=400, score=1.0), budget="concise")
assert out["stats"]["omitted"] > 0
assert "pominięto" in out["prompt"] # użytkownik/model wie, że coś odpadło
def test_empty_report_still_builds_prompt():
out = build_natal_prompt(CHART, {"objects": []})
assert out["stats"]["included"] == 0
assert "brak trafień" in out["prompt"]
assert "# DANE HOROSKOPU" in out["prompt"]
# ------------------------------------------------------------------- okresowy
EVENTS = [
{"technique": "profection", "significator": "Lord of Year: Mars", "start": "2026-04-30",
"exact": "2026-04-30", "end": "2027-04-30", "interpretations_count": 2,
"interpretations": [{"significator": "[Ma", "expanded": "Mars", "effect": "opis marsowy"}]},
{"technique": "solar_arc", "significator": "Sun conj Saturn", "start": "2026-01-01",
"exact": "2026-06-15", "end": "2026-12-31", "interpretations_count": 1,
"interpretations": [{"significator": "[Su [conj [Sa", "expanded": "Sun conjunction Saturn",
"effect": "opis saturniczny"}]},
]
def test_period_prompt_has_dates_and_timeline():
out = build_period_prompt(CHART, EVENTS, "2026-01-01", "2027-01-01")
p = out["prompt"]
assert "2026-01-01 — 2027-01-01" in p
assert "# OŚ CZASU" in p
assert "profection" in p and "solar_arc" in p
assert "2026-06-15" in p # data dokładna zdarzenia
assert "CHRONOLOGICZNIE" in p
assert out["stats"]["events"] == 2
assert out["profile"] == "period"
def test_period_prompt_keeps_timeline_without_interpretations():
"""Padnięta warstwa danych zabiera wskazania, ale NIE oś czasu — ta jest czysto
obliczeniowa i bez niej prognoza okresowa jest bezużyteczna."""
bare = [{k: v for k, v in ev.items()
if k not in ("interpretations", "interpretations_count")} for ev in EVENTS]
out = build_period_prompt(CHART, bare, "2026-01-01", "2027-01-01")
assert out["stats"]["events"] == 2
assert "profection" in out["prompt"] and "2026-06-15" in out["prompt"]
assert out["stats"]["included"] == 0 # brak wskazań, ale oś czasu jest
def test_period_indications_weight_by_hit_count():
items = period_indications(EVENTS)
assert [i["score"] for i in items] == [2.0, 1.0]
def test_natal_indications_flatten_facets():
items = natal_indications(_report(n_samples=4))
assert len(items) == 4
assert all(i["context"].startswith("Sun —") for i in items)
assert all(i["score"] == 2.0 for i in items)
-65
View File
@@ -1,65 +0,0 @@
"""Token międzywarstwowy (LOG-32).
Warstwa logiczna oddaje treść baz, więc musi odrzucać żądania z pominięciem UI.
"""
import pytest
from fastapi import FastAPI
from starlette.testclient import TestClient
from app import security
TOKEN = "tajny-token-testowy"
def _app() -> FastAPI:
app = FastAPI()
security.install(app, "testowa")
@app.get("/health")
def health():
return {"status": "ok"}
@app.get("/secret")
def secret():
return {"rows": ["treść z bazy"]}
return app
@pytest.fixture
def guarded(monkeypatch):
monkeypatch.setenv("INTERNAL_TOKEN", TOKEN)
return TestClient(_app())
@pytest.fixture
def open_app(monkeypatch):
monkeypatch.delenv("INTERNAL_TOKEN", raising=False)
return TestClient(_app())
def test_rejects_request_without_token(guarded):
assert guarded.get("/secret").status_code == 401
def test_rejects_wrong_token(guarded):
r = guarded.get("/secret", headers={security.HEADER: "zly"})
assert r.status_code == 401
assert "treść z bazy" not in r.text
def test_accepts_correct_token(guarded):
r = guarded.get("/secret", headers={security.HEADER: TOKEN})
assert r.status_code == 200
assert r.json()["rows"] == ["treść z bazy"]
def test_health_stays_public(guarded):
"""Sonda k8s nie zna tokenu — /health musi działać bez niego."""
assert guarded.get("/health").status_code == 200
def test_disabled_when_token_unset(open_app):
"""Brak konfiguracji = zgodność wstecz (dev), nie blokada."""
assert not security.enabled()
assert open_app.get("/secret").status_code == 200
-127
View File
@@ -1,127 +0,0 @@
"""Systemy zodiaku (LOG-04): ayanamsy, syderyczny, draconic, RA.
Wartości referencyjne ayanams pochodzą ze **Swiss Ephemeris** (get_ayanamsa_ut)
użytego jako wyrocznia — silnika B nie importujemy tu (izolacja AGPL), więc stałe
są wpięte tak jak referencje astro-seek w innych testach. Nasz model odtwarza je
z dokładnością do ~0,02" w latach 19002100.
"""
import math
import pytest
from app.engine import zodiac as Z
from app.engine.chart import build_chart
from app.engine.formats import sign_index
# wyrocznia: swisseph get_ayanamsa_ut (JD UT -> ayanamsa °)
ORACLE = {
"sidereal_lahiri": {
2451545.0: 23.857092, 2445820.88889: 23.638183,
2415021.0: 22.460550, 2433283.0: 23.158744,
2469808.0: 24.555632, 2488070.0: 25.254287,
},
"sidereal_fagan_bradley": {
2451545.0: 24.740300, 2445820.88889: 24.521391,
2415021.0: 23.343757, 2433283.0: 24.041952,
2469808.0: 25.438840, 2488070.0: 26.137495,
},
"sidereal_krishnamurti": {
2451545.0: 23.760240, 2415021.0: 22.363697,
2469808.0: 24.458780, 2488070.0: 25.157435,
},
}
def _arcsec(a, b):
return abs(a - b) * 3600.0
@pytest.mark.parametrize("name", list(ORACLE))
def test_ayanamsha_matches_swisseph_oracle(name):
for jd, ref in ORACLE[name].items():
got = Z.ayanamsha(name, jd)
assert _arcsec(got, ref) < 0.1, f"{name} @ JD{jd}: {got} vs {ref} ({_arcsec(got, ref):.3f}\")"
def test_julian_day_j2000():
from datetime import datetime, timezone
jd = Z.julian_day(datetime(2000, 1, 1, 12, 0, tzinfo=timezone.utc))
assert abs(jd - 2451545.0) < 1e-6
def test_offset_tropical_is_zero():
assert Z.offset("tropical", 2451545.0) == 0.0
def test_offset_draconic_is_node_longitude():
assert Z.offset("draconic", 2451545.0, node_lon=68.0) == 68.0
# normalizacja
assert Z.offset("draconic", 2451545.0, node_lon=428.0) == pytest.approx(68.0)
def test_offset_draconic_requires_node():
with pytest.raises(ValueError):
Z.offset("draconic", 2451545.0)
def test_unknown_zodiac_rejected():
with pytest.raises(ValueError):
Z.offset("bzdura", 2451545.0)
with pytest.raises(ValueError):
Z.ayanamsha("sidereal_nieistnieje", 2451545.0)
def test_apply_subtracts_and_normalizes():
assert Z.apply(10.0, 20.0) == pytest.approx(350.0)
assert Z.apply(40.0, 24.74) == pytest.approx(15.26)
def test_to_equatorial_reference_points():
# punkt równonocy: λ=0,β=0 -> RA=0, dec=0
ra, dec = Z.to_equatorial(0.0, 0.0, 23.4392911)
assert ra == pytest.approx(0.0, abs=1e-9) and dec == pytest.approx(0.0, abs=1e-9)
# przesilenie letnie: λ=90,β=0 -> RA=90, dec=+ε
ra, dec = Z.to_equatorial(90.0, 0.0, 23.4392911)
assert ra == pytest.approx(90.0, abs=1e-6)
assert dec == pytest.approx(23.4392911, abs=1e-6)
# ---- integracja z pełnym horoskopem ----
def test_sidereal_chart_shifts_signs_back(own_engine, reference_moment):
trop = build_chart(own_engine, reference_moment, "whole_sign")
sid = build_chart(own_engine, reference_moment, "whole_sign", zodiac="sidereal_lahiri")
assert sid["zodiac"] == "sidereal_lahiri"
assert "ayanamsha" in sid and 23.0 < sid["ayanamsha"] < 24.5
tp = {p["name"]: p for p in trop["positions"]}
sp = {p["name"]: p for p in sid["positions"]}
# syderyczna długość = tropikalna ayanamsa (mod 360) dla każdego obiektu
ay = sid["ayanamsha"]
for name in tp:
expect = (tp[name]["decimal"] - ay) % 360.0
assert abs(((sp[name]["decimal"] - expect + 180) % 360) - 180) < 1e-4
def test_sidereal_preserves_house_numbers(own_engine, reference_moment):
# obrót zodiaku nie zmienia numerów domów (geometria niezmiennicza)
trop = build_chart(own_engine, reference_moment, "whole_sign")
sid = build_chart(own_engine, reference_moment, "whole_sign", zodiac="sidereal_lahiri")
tp = {p["name"]: p["house"] for p in trop["positions"]}
sp = {p["name"]: p["house"] for p in sid["positions"]}
assert tp == sp
def test_draconic_puts_node_at_zero_aries(own_engine, reference_moment):
drac = build_chart(own_engine, reference_moment, "whole_sign", zodiac="draconic")
assert drac["zodiac"] == "draconic"
nn = next(p for p in drac["positions"] if p["name"] == "North Node")
# węzeł wznoszący = 0° Barana w draconic
assert sign_index(nn["decimal"]) == 0
assert nn["decimal"] < 0.001 or nn["decimal"] > 359.999
def test_tropical_default_unchanged(own_engine, reference_moment):
# domyślnie tropikalny: brak ayanamshy, zodiac=tropical
chart = build_chart(own_engine, reference_moment, "whole_sign")
assert chart["zodiac"] == "tropical"
assert "ayanamsha" not in chart
@@ -5,44 +5,23 @@ opracowane wyniki. Prezentacja nie sięga bezpośrednio do bazy ani do silnika.
"""
from __future__ import annotations
import os
from typing import Any
import httpx
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]:
"""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())
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)
with httpx.Client(timeout=settings.http_timeout) as client:
r = client.post(f"{self.base_url}/api/query", json=payload)
r.raise_for_status()
return r.json()
def positions(
self,
@@ -52,7 +31,6 @@ class LogicClient:
objects: list[str] | None = None,
house_system: str = "whole_sign",
stations: bool = False,
zodiac: str = "tropical",
) -> dict[str, Any]:
"""Pełny horoskop dla danego momentu — woła logic /chart/positions."""
payload = {
@@ -62,63 +40,22 @@ class LogicClient:
"objects": objects,
"house_system": house_system,
"stations": stations,
"zodiac": zodiac,
}
# stacje wymagają root-findów — dłuższy timeout
timeout = max(settings.http_timeout, 60.0) if stations else settings.http_timeout
return self._post("/chart/positions", payload, timeout)
with httpx.Client(timeout=max(settings.http_timeout, 60.0) if stations else settings.http_timeout) as client:
r = client.post(f"{self.base_url}/chart/positions", json=payload)
r.raise_for_status()
return r.json()
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 prompt(
self, profile: str, when_utc_iso: str, lat: float, lon: float,
budget: str = "medium", from_date: str | None = None, to_date: str | None = None,
provider: str | None = None, model: str | None = None,
) -> dict[str, Any]:
"""Gotowy prompt do LLM z wyliczeń (LOG-29/30) — woła logic /chart/prompt.
`provider` jest potrzebny dla budżetu „maksymalny kontekst modelu": limit
znaków zależy wtedy od okna kontekstu konkretnego modelu."""
payload: dict[str, Any] = {
"profile": profile, "when_utc": when_utc_iso,
"lat": lat, "lon": lon, "budget": budget, "provider": provider,
"model": model,
}
if from_date and to_date:
payload["from_date"], payload["to_date"] = from_date, to_date
return self._post("/chart/prompt", payload, max(settings.http_timeout, 60.0))
def horoscope(
self, profile: str, when_utc_iso: str, lat: float, lon: float,
budget: str = "medium", provider: str | None = None, model: str | None = None,
from_date: str | None = None, to_date: str | None = None,
) -> dict[str, Any]:
"""Napisany horoskop (LOG-31) — woła logic /chart/horoscope.
Generowanie tekstu trwa (zwłaszcza na modelu lokalnym), stąd długi timeout.
"""
payload: dict[str, Any] = {
"profile": profile, "when_utc": when_utc_iso,
"lat": lat, "lon": lon, "budget": budget,
}
if provider:
payload["provider"] = provider
if model:
payload["model"] = model
if from_date and to_date:
payload["from_date"], payload["to_date"] = from_date, to_date
return self._post("/chart/horoscope", payload, max(settings.http_timeout, 300.0))
def llm_models(self) -> dict[str, Any]:
"""Katalog modeli per dostawca (podpowiedzi do pola wyboru w UI)."""
with httpx.Client(timeout=settings.http_timeout) as client:
return link_crypto.call_json(client, "GET", f"{self.base_url}/llm/models",
headers=_auth_headers(), link=_link())
with httpx.Client(timeout=max(settings.http_timeout, 30.0)) as client:
r = client.post(f"{self.base_url}/chart/report", json=payload)
r.raise_for_status()
return r.json()
def timeline(
self, when_utc_iso: str, lat: float, lon: float,
@@ -129,4 +66,7 @@ class LogicClient:
"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))
with httpx.Client(timeout=max(settings.http_timeout, 60.0)) as client:
r = client.post(f"{self.base_url}/chart/timeline", json=payload)
r.raise_for_status()
return r.json()
-466
View File
@@ -1,466 +0,0 @@
"""Szyfrowanie łączy między warstwami (PRE-16 / LOG-33).
Do tej pory warstwy rozmawiały ze sobą zwykłym HTTP-em wewnątrz klastra. Token
międzywarstwowy (LOG-32) mówił KTO pyta, ale nie ukrywał CZEGO dotyczy odpowiedź
— a płyną nią surowe wiersze oryginalnych baz interpretacyjnych, czyli rdzeń
produktu. Kto podsłuchał ruch wewnątrz sieci (drugi pod, port mirror na switchu,
zrzut z węzła), miał je w całości.
Ten moduł zamyka tę drogę: **AES-256-GCM** na ciele każdego żądania i odpowiedzi.
GCM daje jednocześnie poufność i uwierzytelnienie — cudzy albo podmieniony bajt
nie odszyfruje się w ogóle, więc nie ma osobnego problemu „zaszyfrowane, ale
podatne na modyfikację".
**Dwa niezależne klucze**, po jednym na parę rozmówców:
* ``LINK_KEY_PRESENTATION_LOGIC`` — prezentacja ↔ logika,
* ``LINK_KEY_LOGIC_DATA`` — logika ↔ dane.
Dzięki temu przejęcie klucza prezentacji nie daje dostępu do warstwy danych,
gdzie leżą całe bazy. Logika trzyma oba, bo rozmawia w obie strony.
Z każdego klucza łącza wyprowadzamy **osobne podklucze na kierunek** (HKDF).
Żądanie i odpowiedź nigdy nie szyfrują się tym samym kluczem, więc powtórzenie
losowej jednorazówki w jedną stronę nie osłabia drugiej.
Format ramki (bo strumień odpowiedzi może iść kawałkami — patrz okno postępu):
[4 bajty długości][magia "AL1"][12 bajtów jednorazówki][szyfrogram + znacznik]
Do materiału uwierzytelnianego (AAD) wchodzą kierunek, ścieżka, znacznik czasu
i numer ramki. Skutek: ramki nie da się przekleić do innego endpointu, odtworzyć
po czasie (dopuszczalny poślizg ``MAX_SKEW``) ani przestawić w strumieniu.
Bez ustawionego klucza moduł **przepuszcza ruch otwartym tekstem** (dev, zgodność
wstecz) i krzyczy o tym przy starcie. Gdy klucz JEST ustawiony, warstwa serwerowa
działa fail-closed: nieszyfrowane żądanie dostaje odmowę, żeby przypadkowa
regresja po stronie klienta nie oznaczała cichego powrotu do jawnego ruchu.
"""
from __future__ import annotations
import base64
import binascii
import logging
import os
import struct
import time
from typing import Iterable, Iterator
from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
log = logging.getLogger("astrololo.link")
MAGIC = b"AL1"
VERSION = "v1"
NONCE_BYTES = 12
KEY_BYTES = 32 # AES-256
LENGTH_PREFIX = 4
MAX_FRAME = 64 * 1024 * 1024 # zapora przed alokacją z podanej długości
MAX_SKEW_SECONDS = 300.0
HEADER_ENC = "X-Astrololo-Enc"
HEADER_TS = "X-Astrololo-Enc-Ts"
CONTENT_TYPE = "application/vnd.astrololo.enc"
ENV_PRESENTATION_LOGIC = "LINK_KEY_PRESENTATION_LOGIC"
ENV_LOGIC_DATA = "LINK_KEY_LOGIC_DATA"
ENV_REQUIRED = "LINK_ENCRYPTION_REQUIRED"
REQUEST, RESPONSE = b"req", b"res"
# Sondy k8s pukają tu bez klucza i tak ma zostać — inaczej pierwsza literówka
# w sekrecie kładłaby pody zamiast pokazać błąd w aplikacji.
PUBLIC_PATHS = frozenset({"/health"})
class LinkError(Exception):
"""Cokolwiek poszło nie tak z kopertą — celowo bez szczegółów na zewnątrz."""
# --------------------------------------------------------------------- klucze
def parse_key(raw: str) -> bytes:
"""Klucz z konfiguracji: hex (64 znaki) albo base64. Zawsze 32 bajty."""
text = raw.strip()
if not text:
raise LinkError("pusty klucz łącza")
try:
key = bytes.fromhex(text)
except ValueError:
try:
key = base64.b64decode(text, validate=True)
except (binascii.Error, ValueError) as exc:
raise LinkError("klucz łącza nie jest ani hexem, ani base64") from exc
if len(key) != KEY_BYTES:
raise LinkError(
f"klucz łącza ma {len(key)} B zamiast {KEY_BYTES} — wygeneruj przez "
f"`openssl rand -hex 32`"
)
return key
def key_from_env(env_name: str) -> bytes | None:
"""Klucz albo None. Zły klucz to wyjątek OD RAZU — nie przy pierwszym żądaniu."""
raw = os.getenv(env_name, "")
return parse_key(raw) if raw.strip() else None
def encryption_required() -> bool:
"""Czy brak klucza ma być błędem, a nie cichym powrotem do jawnego ruchu.
Serwer sam z siebie broni się fail-closed, ale to za mało: klient BEZ klucza
wysyła pytanie otwartym tekstem i dopiero potem dostaje odmowę — czyli treść
zapytania zdążyła już przelecieć przez sieć. Ta flaga zatrzymuje go, zanim
cokolwiek opuści proces. Ustawiana razem z kluczami we wdrożeniu.
"""
return os.getenv(ENV_REQUIRED, "").strip().lower() in {"1", "true", "yes", "on"}
def _subkey(link_key: bytes, direction: bytes) -> bytes:
return HKDF(
algorithm=hashes.SHA256(), length=KEY_BYTES, salt=None,
info=b"astrololo/link/" + direction,
).derive(link_key)
class Link:
"""Jedna para rozmówców: klucz plus wyprowadzone z niego podklucze."""
def __init__(self, link_key: bytes) -> None:
self._by_direction = {
REQUEST: AESGCM(_subkey(link_key, REQUEST)),
RESPONSE: AESGCM(_subkey(link_key, RESPONSE)),
}
# ---------------------------------------------------------- pojedyncza ramka
def _aad(self, direction: bytes, path: str, stamp: str, seq: int) -> bytes:
return b"|".join([MAGIC, direction, path.encode("utf-8"),
stamp.encode("ascii"), str(seq).encode("ascii")])
def seal(self, direction: bytes, path: str, stamp: str, seq: int,
plaintext: bytes) -> bytes:
nonce = os.urandom(NONCE_BYTES)
sealed = self._by_direction[direction].encrypt(
nonce, plaintext, self._aad(direction, path, stamp, seq))
return MAGIC + nonce + sealed
def open(self, direction: bytes, path: str, stamp: str, seq: int,
frame: bytes) -> bytes:
if not frame.startswith(MAGIC):
raise LinkError("ramka bez znacznika astrololo")
body = frame[len(MAGIC):]
if len(body) <= NONCE_BYTES:
raise LinkError("ramka za krótka")
nonce, sealed = body[:NONCE_BYTES], body[NONCE_BYTES:]
try:
return self._by_direction[direction].decrypt(
nonce, sealed, self._aad(direction, path, stamp, seq))
except InvalidTag as exc:
# Jeden komunikat na wszystkie przypadki: zły klucz, podmieniony bajt,
# przeklejenie z innej ścieżki, przestawiona ramka. Rozróżnianie ich
# na zewnątrz podpowiadałoby atakującemu, w co trafił.
raise LinkError("nie udało się odszyfrować — zły klucz albo naruszone dane") from exc
# ------------------------------------------------------------ strumień ramek
def seal_stream(self, direction: bytes, path: str, stamp: str,
chunks: Iterable[bytes]) -> Iterator[bytes]:
for seq, chunk in enumerate(chunks):
yield frame_out(self.seal(direction, path, stamp, seq, chunk))
def open_stream(self, direction: bytes, path: str, stamp: str,
raw: bytes) -> Iterator[bytes]:
for seq, frame in enumerate(frames_in(raw)):
yield self.open(direction, path, stamp, seq, frame)
def open_all(self, direction: bytes, path: str, stamp: str, raw: bytes) -> bytes:
return b"".join(self.open_stream(direction, path, stamp, raw))
# ---------------------------------------------------------------- ramkowanie
def frame_out(payload: bytes) -> bytes:
return struct.pack(">I", len(payload)) + payload
def frames_in(raw: bytes) -> Iterator[bytes]:
"""Rozbiera bufor na ramki. Ucięty strumień to błąd, nie cicha strata danych."""
offset = 0
while offset < len(raw):
if offset + LENGTH_PREFIX > len(raw):
raise LinkError("urwana ramka (brak nagłówka długości)")
(size,) = struct.unpack(">I", raw[offset:offset + LENGTH_PREFIX])
if size > MAX_FRAME:
raise LinkError("ramka ponad dopuszczalny rozmiar")
offset += LENGTH_PREFIX
if offset + size > len(raw):
raise LinkError("urwana ramka (za mało danych)")
yield raw[offset:offset + size]
offset += size
def unframe_incremental(buffer: bytearray) -> Iterator[bytes]:
"""Wyjmuje z bufora KOMPLETNE ramki i zjada je; resztę zostawia na później.
Dla odbioru na żywo: kawałki przychodzą podzielone dowolnie i ramka potrafi
rozjechać się między dwa odczyty.
"""
while True:
if len(buffer) < LENGTH_PREFIX:
return
(size,) = struct.unpack(">I", buffer[:LENGTH_PREFIX])
if size > MAX_FRAME:
raise LinkError("ramka ponad dopuszczalny rozmiar")
if len(buffer) < LENGTH_PREFIX + size:
return
frame = bytes(buffer[LENGTH_PREFIX:LENGTH_PREFIX + size])
del buffer[:LENGTH_PREFIX + size]
yield frame
# ------------------------------------------------------------- świeżość ruchu
def stamp_now() -> str:
return f"{time.time():.3f}"
def check_stamp(stamp: str) -> None:
"""Odrzuca ramki spoza okna czasowego — inaczej podsłuchane żądanie dałoby się
odtworzyć w dowolnym momencie w przyszłości."""
try:
sent = float(stamp)
except (TypeError, ValueError) as exc:
raise LinkError("brak albo błędny znacznik czasu") from exc
if abs(time.time() - sent) > MAX_SKEW_SECONDS:
raise LinkError("znacznik czasu poza dopuszczalnym oknem")
# =========================================================== strona serwerowa
class LinkCryptoMiddleware:
"""Rozszyfrowuje wchodzące żądania i zaszyfrowuje wychodzące odpowiedzi.
Napisane jako czyste ASGI, nie ``@app.middleware("http")``, bo trzeba
podmienić CIAŁO żądania jeszcze zanim zobaczy je FastAPI, oraz przepuścić
odpowiedź strumieniową kawałek po kawałku, bez zbierania jej w pamięci.
"""
def __init__(self, app, link: Link | None, layer: str) -> None:
self.app = app
self.link = link
self.layer = layer
async def __call__(self, scope, receive, send):
if scope["type"] != "http" or self.link is None or scope["path"] in PUBLIC_PATHS:
return await self.app(scope, receive, send)
path = scope["path"]
headers = {k.decode("latin-1").lower(): v.decode("latin-1") for k, v in scope["headers"]}
if headers.get(HEADER_ENC.lower()) != VERSION:
# Fail-closed. Klucz jest ustawiony, więc jawne żądanie oznacza albo
# pomyłkę w konfiguracji, albo kogoś obcego — w obu wypadkach nie
# chcemy po cichu wrócić do jawnego ruchu.
log.warning("warstwa %s: odrzucone żądanie bez szyfrowania łącza (%s)",
self.layer, path)
return await _refuse(send, "Łącze międzywarstwowe wymaga szyfrowania.")
stamp = headers.get(HEADER_TS.lower(), "")
try:
check_stamp(stamp)
plaintext = self.link.open_all(REQUEST, path, stamp, await _read_body(receive))
except LinkError as exc:
log.warning("warstwa %s: %s (%s)", self.layer, exc, path)
return await _refuse(send, "Nie udało się odczytać zaszyfrowanego żądania.")
scope = dict(scope)
scope["headers"] = _rewritten_headers(scope["headers"], len(plaintext))
await self.app(scope, _replay(plaintext, receive), self._sealing_send(send, path))
def _sealing_send(self, send, path: str):
state: dict = {"stamp": "", "seq": 0}
async def sealing(message):
if message["type"] == "http.response.start":
state["stamp"] = stamp_now()
keep = [(k, v) for k, v in message.get("headers", [])
if k.lower() not in (b"content-length", b"content-type")]
message = dict(message)
message["headers"] = keep + [
(b"content-type", CONTENT_TYPE.encode()),
(HEADER_ENC.lower().encode(), VERSION.encode()),
(HEADER_TS.lower().encode(), state["stamp"].encode()),
]
return await send(message)
if message["type"] == "http.response.body":
chunk = message.get("body", b"")
sealed = b""
if chunk:
sealed = frame_out(self.link.seal(
RESPONSE, path, state["stamp"], state["seq"], chunk))
state["seq"] += 1
return await send({"type": "http.response.body", "body": sealed,
"more_body": message.get("more_body", False)})
return await send(message)
return sealing
def _rewritten_headers(raw: Iterable[tuple[bytes, bytes]], length: int):
"""Po odszyfrowaniu ciało ma inną długość i zwykły typ — inaczej FastAPI
próbowałby sparsować JSON o cudzej deklarowanej wielkości."""
kept = [(k, v) for k, v in raw if k.lower() not in (b"content-length", b"content-type")]
kept.append((b"content-length", str(length).encode()))
if length:
kept.append((b"content-type", b"application/json"))
return kept
async def _read_body(receive) -> bytes:
body = bytearray()
while True:
message = await receive()
if message["type"] == "http.disconnect":
raise LinkError("rozłączenie w trakcie odbioru żądania")
body += message.get("body", b"")
if not message.get("more_body", False):
return bytes(body)
def _replay(body: bytes, original):
"""Podstawia odszyfrowane ciało jako jedyną porcję wejścia dla aplikacji.
Po oddaniu ciała oddajemy głos ORYGINALNEMU `receive`, zamiast od razu
zgłaszać rozłączenie. Odpowiedź strumieniowa nasłuchuje bowiem rozłączenia
równolegle do wysyłania i przerywa się, gdy je zobaczy — na skróconej wersji
okno postępu dostawało pustą odpowiedź, choć zwykłe żądania działały.
"""
delivered = False
async def receive():
nonlocal delivered
if delivered:
return await original()
delivered = True
return {"type": "http.request", "body": body, "more_body": False}
return receive
async def _refuse(send, detail: str) -> None:
"""Odmowa leci JAWNIE — rozmówca właśnie pokazał, że nie umie odszyfrować,
więc zaszyfrowany komunikat o błędzie byłby dla niego nieczytelny."""
payload = f'{{"detail":"{detail}"}}'.encode("utf-8")
await send({"type": "http.response.start", "status": 400, "headers": [
(b"content-type", b"application/json"),
(b"content-length", str(len(payload)).encode()),
]})
await send({"type": "http.response.body", "body": payload})
def install(app, env_name: str, layer: str):
"""Podpina szyfrowanie łącza. Wołać PO `security.install`, żeby także odmowa
tokenowa (401) wracała zaszyfrowana — inaczej klient by jej nie odczytał."""
link_key = key_from_env(env_name)
if link_key is None and encryption_required():
# Celowo wywracamy start. Ta sama zasada co przy sekrecie logowania:
# wolimy widoczną awarię niż usługę, która wstała i po cichu nie chroni
# niczego. Pod w CrashLoop widać od razu, jawny ruch — nie.
raise LinkError(
f"{ENV_REQUIRED} jest włączone, ale {env_name} nie ustawiony — "
f"warstwa {layer} nie wystartuje bez klucza łącza"
)
if link_key is None:
log.warning(
"UWAGA: %s nie ustawiony — warstwa %s rozmawia z sąsiadem JAWNYM tekstem, "
"więc treść baz interpretacyjnych jest widoczna dla każdego, kto podsłucha "
"ruch wewnątrz sieci.", env_name, layer,
)
return None
link = Link(link_key)
app.add_middleware(LinkCryptoMiddleware, link=link, layer=layer)
log.info("warstwa %s: łącze szyfrowane (AES-256-GCM, klucz z %s)", layer, env_name)
return link
# ============================================================ strona kliencka
def call(client, method: str, url: str, *, payload=None,
headers: dict[str, str] | None = None, link: Link | None) -> bytes:
"""Żądanie do sąsiedniej warstwy; zwraca odszyfrowane ciało odpowiedzi.
Ścieżkę do materiału uwierzytelnianego bierzemy Z URL-a, a nie z osobnego
argumentu — gdyby klient i serwer liczyły ją inaczej, każde żądanie kończyłoby
się niejasnym błędem odszyfrowania.
"""
import json as _json
import httpx
request_headers = dict(headers or {})
if link is None:
if encryption_required():
# Zatrzymujemy się PRZED wysłaniem. Gdyby polecieć jawnie i dopiero
# zebrać odmowę, pytanie byłoby już na kablu — a to właśnie ono niesie
# sygnifikatory, o które pytamy bazę.
raise LinkError(
f"{ENV_REQUIRED} jest włączone, ale brak klucza łącza — żądanie "
f"NIE zostało wysłane, żeby jego treść nie poszła jawnym tekstem"
)
response = client.request(method, url, json=payload, headers=request_headers)
response.raise_for_status()
return response.content
path = httpx.URL(url).path
stamp = stamp_now()
plaintext = b"" if payload is None else _json.dumps(payload).encode("utf-8")
body = frame_out(link.seal(REQUEST, path, stamp, 0, plaintext))
request_headers.update({HEADER_ENC: VERSION, HEADER_TS: stamp,
"Content-Type": CONTENT_TYPE})
response = client.request(method, url, content=body, headers=request_headers)
if response.status_code >= 400 and response.headers.get(HEADER_ENC) != VERSION:
log.error("łącze %s odmówiło: %s", path, response.text[:200])
response.raise_for_status()
if response.headers.get(HEADER_ENC) != VERSION:
raise LinkError("odpowiedź przyszła nieszyfrowana, choć klucz łącza jest ustawiony")
reply_stamp = response.headers.get(HEADER_TS, "")
check_stamp(reply_stamp)
return link.open_all(RESPONSE, path, reply_stamp, response.content)
def call_json(client, method: str, url: str, *, payload=None,
headers: dict[str, str] | None = None, link: Link | None):
import json as _json
return _json.loads(call(client, method, url, payload=payload,
headers=headers, link=link))
def open_response_stream(response, link: Link | None) -> Iterator[bytes]:
"""Odbiór odpowiedzi płynącej kawałkami (okno postępu).
Ramka potrafi rozjechać się między dwa odczyty z gniazda, więc składamy ją
w buforze zamiast zakładać, że każdy kawałek to komplet.
"""
if link is None:
yield from response.iter_bytes()
return
if response.headers.get(HEADER_ENC) != VERSION:
raise LinkError("strumień przyszedł nieszyfrowany, choć klucz łącza jest ustawiony")
stamp = response.headers.get(HEADER_TS, "")
check_stamp(stamp)
path = response.request.url.path
buffer = bytearray()
seq = 0
for chunk in response.iter_bytes():
buffer += chunk
for frame in unframe_incremental(buffer):
yield link.open(RESPONSE, path, stamp, seq, frame)
seq += 1
if buffer:
raise LinkError("strumień urwał się w połowie ramki")
+10 -61
View File
@@ -17,7 +17,7 @@ from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from app import geocode, security
from app import geocode
from app.clients.logic_client import LogicClient
from app.config import DEFAULT_LOCATION_LABEL, default_form
@@ -25,7 +25,6 @@ app = FastAPI(title="astrololo · warstwa prezentacji")
app.mount("/static", StaticFiles(directory="app/static"), name="static")
templates = Jinja2Templates(directory="app/templates")
logic = LogicClient()
security.install(app) # logowanie + limit żądań (LOG-32)
def _build_utc(date: str, time: str, tz_offset: float) -> tuple[str, str]:
@@ -39,15 +38,6 @@ def _build_utc(date: str, time: str, tz_offset: float) -> tuple[str, str]:
return utc.isoformat(), label
def _llm_catalog() -> dict:
"""Podpowiedzi modeli dla pola wyboru. Awaria logiki nie może wywrócić strony —
pole modelu jest tekstowe, więc bez katalogu nadal da się wpisać model ręcznie."""
try:
return logic.llm_models()
except httpx.HTTPError:
return {"providers": {}, "defaults": {}}
def _logic_error(e: Exception) -> str:
if isinstance(e, httpx.HTTPStatusError) and e.response.status_code == 404:
return (
@@ -62,8 +52,7 @@ def _logic_error(e: Exception) -> str:
def chart_form(request: Request):
return templates.TemplateResponse(
request, "chart.html",
{"result": None, "form": default_form(), "location_label": DEFAULT_LOCATION_LABEL,
"llm_catalog": _llm_catalog()},
{"result": None, "form": default_form(), "location_label": DEFAULT_LOCATION_LABEL},
)
@@ -77,18 +66,16 @@ def chart_compute(
lon: float = Form(0.0),
house_system: str = Form("whole_sign"),
stations: bool = Form(False),
zodiac: str = Form("tropical"),
):
form = {"date": date, "time": time, "tz_offset": tz_offset,
"lat": lat, "lon": lon, "house_system": house_system, "stations": stations,
"zodiac": zodiac}
"lat": lat, "lon": lon, "house_system": house_system, "stations": stations}
ctx: dict = {"form": form, "result": None, "error": None, "moment": None}
try:
iso_utc, label = _build_utc(date, time, tz_offset)
ctx["moment"] = label
ctx["result"] = logic.positions(
when_utc_iso=iso_utc, lat=lat, lon=lon,
house_system=house_system, stations=stations, zodiac=zodiac,
house_system=house_system, stations=stations,
)
except (httpx.HTTPError,) as e:
ctx["error"] = _logic_error(e)
@@ -125,8 +112,7 @@ def significators_search(
def interpret_form(request: Request):
return templates.TemplateResponse(
request, "interpret.html",
{"result": None, "form": default_form(), "location_label": DEFAULT_LOCATION_LABEL,
"llm_catalog": _llm_catalog()},
{"result": None, "form": default_form(), "location_label": DEFAULT_LOCATION_LABEL},
)
@@ -139,30 +125,13 @@ def interpret_run(
lat: float = Form(0.0),
lon: float = Form(0.0),
group: bool = Form(False),
action: str = Form("report"),
prompt_budget: str = Form("medium"),
llm_provider: str = Form("local"),
llm_model: str = Form(""),
):
form = {"date": date, "time": time, "tz_offset": tz_offset,
"lat": lat, "lon": lon, "group": group, "prompt_budget": prompt_budget,
"llm_provider": llm_provider, "llm_model": llm_model}
ctx: dict = {"form": form, "result": None, "error": None, "moment": None,
"llm_catalog": _llm_catalog()}
"lat": lat, "lon": lon, "group": group}
ctx: dict = {"form": form, "result": None, "error": None, "moment": None}
try:
iso_utc, label = _build_utc(date, time, tz_offset)
ctx["moment"] = label
if action == "prompt":
ctx["prompt_result"] = logic.prompt(
profile="natal", when_utc_iso=iso_utc, lat=lat, lon=lon, budget=prompt_budget,
provider=llm_provider, model=llm_model,
)
elif action == "horoscope":
ctx["prompt_result"] = logic.horoscope(
profile="natal", when_utc_iso=iso_utc, lat=lat, lon=lon,
budget=prompt_budget, provider=llm_provider, model=llm_model,
)
else:
ctx["result"] = logic.report(when_utc_iso=iso_utc, lat=lat, lon=lon, group=group)
except httpx.HTTPError as e:
ctx["error"] = _logic_error(e)
@@ -176,8 +145,7 @@ def interpret_run(
def timeline_form(request: Request):
return templates.TemplateResponse(
request, "timeline.html",
{"result": None, "form": default_form(), "location_label": DEFAULT_LOCATION_LABEL,
"llm_catalog": _llm_catalog()},
{"result": None, "form": default_form(), "location_label": DEFAULT_LOCATION_LABEL},
)
@@ -191,32 +159,13 @@ def timeline_run(
lon: float = Form(0.0),
from_date: str = Form(...),
to_date: str = Form(...),
action: str = Form("timeline"),
prompt_budget: str = Form("medium"),
llm_provider: str = Form("local"),
llm_model: str = Form(""),
):
form = {"date": date, "time": time, "tz_offset": tz_offset, "lat": lat, "lon": lon,
"from_date": from_date, "to_date": to_date, "prompt_budget": prompt_budget,
"llm_provider": llm_provider, "llm_model": llm_model}
ctx: dict = {"form": form, "result": None, "error": None, "moment": None,
"llm_catalog": _llm_catalog()}
"from_date": from_date, "to_date": to_date}
ctx: dict = {"form": form, "result": None, "error": None, "moment": None}
try:
iso_utc, label = _build_utc(date, time, tz_offset)
ctx["moment"] = label
if action == "prompt":
ctx["prompt_result"] = logic.prompt(
profile="period", when_utc_iso=iso_utc, lat=lat, lon=lon,
budget=prompt_budget, from_date=from_date, to_date=to_date,
provider=llm_provider, model=llm_model,
)
elif action == "horoscope":
ctx["prompt_result"] = logic.horoscope(
profile="period", when_utc_iso=iso_utc, lat=lat, lon=lon,
budget=prompt_budget, provider=llm_provider, model=llm_model,
from_date=from_date, to_date=to_date,
)
else:
ctx["result"] = logic.timeline(
when_utc_iso=iso_utc, lat=lat, lon=lon,
from_date=from_date, to_date=to_date, interpret=True,
-151
View File
@@ -1,151 +0,0 @@
"""Ochrona dostępu do aplikacji (LOG-32).
Rdzeniem produktu są oryginalne bazy interpretacyjne. Aplikacja podaje ich treść
na wielu ścieżkach (`/significators`, `/interpret`, generator promptu), więc BRAK
uwierzytelnienia oznacza, że każdy w sieci może je wypompować — bez udziału
jakiegokolwiek modelu językowego. Ten moduł zamyka tę drogę.
Dwa mechanizmy:
* **HTTP Basic** — wejście do aplikacji; włącza się, gdy ustawiono APP_PASSWORD.
* **limit żądań** — hamuje masowe odpytywanie (eksfiltrację przez pętlę zapytań);
rozliczany per adres klienta, a za odwrotnym proxy — po TRUST_PROXY=true —
per adres z nagłówka, nie per adres proxy (patrz `client_ip`).
Świadomie NIE logujemy treści żądań ani promptów — logi to kolejny nośnik wycieku.
UWAGA: bez APP_PASSWORD ochrona jest WYŁĄCZONA (zgodność wstecz i wygoda dev).
Wtedy przy starcie leci głośne ostrzeżenie — żeby nikt nie wdrożył tego w
przekonaniu, że jest chroniony.
"""
from __future__ import annotations
import base64
import binascii
import logging
import os
import secrets
import time
from collections import deque
from fastapi import Request
from fastapi.responses import HTMLResponse, JSONResponse
log = logging.getLogger("astrololo.security")
MAX_TRACKED_CLIENTS_DEFAULT = 4096
MAX_TRACKED_CLIENTS = MAX_TRACKED_CLIENTS_DEFAULT # zabezpieczenie przed puchnięciem pamięci
# konfiguracja czytana leniwie — testy i restart mogą ją zmienić bez importu modułu
def app_user() -> str:
return os.getenv("APP_USER", "astrololo")
def app_password() -> str:
return os.getenv("APP_PASSWORD", "")
def rate_limit_per_min() -> int:
return int(os.getenv("RATE_LIMIT_PER_MIN", "120"))
def trust_proxy() -> bool:
return os.getenv("TRUST_PROXY", "").strip().lower() in {"1", "true", "yes", "on"}
PUBLIC_PATHS = frozenset({"/health"})
PUBLIC_PREFIXES = ("/static/",)
_hits: dict[str, deque[float]] = {}
def auth_enabled() -> bool:
return bool(app_password())
def _is_public(path: str) -> bool:
return path in PUBLIC_PATHS or path.startswith(PUBLIC_PREFIXES)
def _authorized(header: str | None) -> bool:
if not header or not header.lower().startswith("basic "):
return False
try:
raw = base64.b64decode(header.split(" ", 1)[1]).decode("utf-8")
user, _, password = raw.partition(":")
except (binascii.Error, UnicodeDecodeError, IndexError):
return False
# porównanie odporne na atak czasowy; oba pola muszą się zgadzać
ok_user = secrets.compare_digest(user, app_user())
ok_pass = secrets.compare_digest(password, app_password())
return ok_user and ok_pass
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
def install(app) -> None:
"""Podpina ochronę pod wszystkie ścieżki poza /health i /static."""
if not auth_enabled():
log.warning(
"UWAGA: APP_PASSWORD nie ustawione — aplikacja jest OTWARTA dla każdego, "
"kto ma do niej dostęp sieciowy, wraz z treścią baz interpretacyjnych."
)
@app.middleware("http")
async def _guard(request: Request, call_next):
if _is_public(request.url.path):
return await call_next(request)
if _rate_limited(client_ip(request)):
return JSONResponse(
{"detail": "Zbyt wiele żądań — spróbuj za chwilę."},
status_code=429, headers={"Retry-After": "60"},
)
if auth_enabled() and not _authorized(request.headers.get("authorization")):
return HTMLResponse(
"<h1>401 — wymagane logowanie</h1>", status_code=401,
headers={"WWW-Authenticate": 'Basic realm="astrololo"'},
)
return await call_next(request)
-31
View File
@@ -1,31 +0,0 @@
// Kopiowanie do schowka dla przycisków [data-copy="#selektor"].
// clipboard API wymaga secure context (https/localhost), a aplikacja bywa serwowana
// po http://<ip> — dlatego jest awaryjne przejście na zaznaczenie + execCommand.
document.addEventListener('click', function (e) {
const btn = e.target.closest('[data-copy]');
if (!btn) return;
const src = document.querySelector(btn.getAttribute('data-copy'));
if (!src) return;
const done = ok => {
const label = btn.textContent;
btn.textContent = ok ? 'Skopiowano ✓' : 'Nie udało się — zaznacz i skopiuj ręcznie';
setTimeout(() => { btn.textContent = label; }, 2500);
};
const text = src.value !== undefined ? src.value : src.textContent;
if (navigator.clipboard && window.isSecureContext) {
navigator.clipboard.writeText(text).then(() => done(true), () => done(false));
return;
}
// awaryjnie: zaznacz zawartość i spróbuj starym poleceniem
try {
src.focus();
if (src.select) src.select();
done(document.execCommand('copy'));
} catch (err) {
done(false);
}
});
@@ -1,58 +0,0 @@
// Podpowiedzi modeli zależne od wybranego dostawcy.
//
// Pole modelu jest CELOWO tekstowe (input + datalist), a nie zamkniętym <select>:
// katalog to tylko wygoda, a konto może mieć dostęp do modeli, o których kod nie
// wie. Puste pole = model domyślny dostawcy.
document.addEventListener('DOMContentLoaded', function () {
const providerEl = document.getElementById('llmProvider');
const modelEl = document.getElementById('llmModel');
const listEl = document.getElementById('llmModelList');
const hintEl = document.getElementById('llmModelHint');
const dataEl = document.getElementById('llmCatalog');
if (!providerEl || !modelEl || !listEl || !dataEl) return;
let catalog = {};
let defaults = {};
try {
const parsed = JSON.parse(dataEl.textContent || '{}');
catalog = parsed.providers || {};
defaults = parsed.defaults || {};
} catch (e) {
return; // brak katalogu — pole nadal działa jako wolny tekst
}
const fmt = n =>
n >= 1000000 ? (n / 1000000) + 'M' : (n >= 1000 ? Math.round(n / 1000) + 'k' : String(n));
function refresh(resetValue) {
const provider = providerEl.value;
const models = catalog[provider] || [];
listEl.innerHTML = '';
models.forEach(m => {
const opt = document.createElement('option');
opt.value = m.id;
opt.label = m.label + ' · kontekst ' + fmt(m.context_window);
listEl.appendChild(opt);
});
const fallback = defaults[provider] || '';
modelEl.placeholder = fallback ? 'domyślny: ' + fallback : 'domyślny dostawcy';
// po zmianie dostawcy stary model nie ma sensu (np. gpt-4o u Anthropica)
if (resetValue) modelEl.value = '';
if (hintEl) {
const chosen = models.find(m => m.id === modelEl.value.trim());
hintEl.textContent = chosen
? chosen.label + ' — okno kontekstu ' + fmt(chosen.context_window) +
' tokenów, maksymalna odpowiedź ' + fmt(chosen.max_output) + '.'
: 'Zostaw puste, by użyć modelu domyślnego. Możesz też wpisać dowolny ' +
'identyfikator modelu, do którego Twoje konto ma dostęp.';
}
}
providerEl.addEventListener('change', () => refresh(true));
modelEl.addEventListener('input', () => refresh(false));
refresh(false);
});
@@ -81,10 +81,3 @@ tr:last-child td { border-bottom: none; }
.geo-pin { font-size: 20px; line-height: 24px; text-align: center; }
/* kafelki OSM są jasne — trzymamy kontrolki czytelne na ciemnym tle */
.leaflet-control-attribution { font-size: .68rem; }
/* Generator promptu do LLM (LOG-29/30) */
textarea.prompt { width: 100%; margin-top: .5rem; padding: .7rem .8rem; box-sizing: border-box;
background: #12132a; color: var(--ink); border: 1px solid var(--line);
border-radius: 10px; font-family: ui-monospace, SFMono-Regular, Menlo, monospace;
font-size: .82rem; line-height: 1.45; resize: vertical; white-space: pre; }
textarea.prompt:focus { outline: 2px solid var(--accent); outline-offset: 1px; }
@@ -1,115 +0,0 @@
{# Katalog modeli wstrzykiwany przez handler — pole jest tekstowe, więc
można wpisać dowolny model, do którego konto ma dostęp. #}
<script type="application/json" id="llmCatalog">{{ llm_catalog | tojson }}</script>
{# Wspólny blok: sterowanie generowaniem + prompt (LOG-29/30) + horoskop (LOG-31).
Używany na ekranach Interpretacje i Kalendarz. #}
<div class="opts">
<label>Budżet promptu
<select name="prompt_budget">
{% set pb = form.prompt_budget or 'medium' %}
<option value="concise" {{ 'selected' if pb == 'concise' else '' }}>zwięzły (~4 tys. znaków)</option>
<option value="medium" {{ 'selected' if pb == 'medium' else '' }}>średni (~12 tys.)</option>
<option value="extensive" {{ 'selected' if pb == 'extensive' else '' }}>obszerny (~30 tys.)</option>
<option value="huge" {{ 'selected' if pb == 'huge' else '' }}>bardzo obszerny (~120 tys.)</option>
<option value="max" {{ 'selected' if pb == 'max' else '' }}>maksymalny kontekst modelu</option>
</select>
</label>
<label>Dostawca
<select name="llm_provider" id="llmProvider">
{% set lp = form.llm_provider or 'local' %}
<option value="local" {{ 'selected' if lp == 'local' else '' }}>lokalny — nic nie opuszcza sieci</option>
<option value="anthropic" {{ 'selected' if lp == 'anthropic' else '' }}>Anthropic — dane wychodzą</option>
<option value="openai" {{ 'selected' if lp == 'openai' else '' }}>OpenAI — dane wychodzą</option>
</select>
</label>
<label>Model
<input type="text" name="llm_model" id="llmModel" list="llmModelList"
value="{{ form.llm_model or '' }}" placeholder="domyślny dostawcy"
autocomplete="off">
<datalist id="llmModelList"></datalist>
</label>
</div>
<p class="muted small" id="llmModelHint"></p>
<p class="muted small">
Prompt zawiera <strong>oryginalne opisy z baz</strong> oraz dane urodzeniowe. Model lokalny
przetwarza je u nas; wybór dostawcy w chmurze oznacza, że ta treść <strong>opuszcza naszą
sieć</strong>. Możesz najpierw obejrzeć prompt, a dopiero potem wysłać.
</p>
{% if prompt_result %}
{% set st = prompt_result.stats %}
{# --- wynik: napisany horoskop (LOG-31) + transparentność (PRE-15) --- #}
{% if prompt_result.horoscope %}
<div class="meta">
Horoskop napisany przez: <strong>{{ prompt_result.provider }}</strong> ·
model: {{ prompt_result.model }}
{% if prompt_result.usage and prompt_result.usage.completion_tokens %}
· tokeny odpowiedzi: {{ prompt_result.usage.completion_tokens }}
{% endif %}
{% if prompt_result.leaves_lan %}
· <strong class="retro">dane opuściły sieć</strong>
{% else %}
· dane nie opuściły sieci
{% endif %}
</div>
<div class="actions">
<button type="button" class="ghost" data-copy="#horoscopeText">Kopiuj horoskop</button>
</div>
<textarea id="horoscopeText" class="prompt" rows="20" readonly>{{ prompt_result.horoscope }}</textarea>
<p class="muted small">
Treść wygenerował model językowy na podstawie {{ st.included }} wskazań z baz.
<strong>Nie stanowi porady medycznej, prawnej ani finansowej.</strong>
</p>
{% endif %}
{% if prompt_result.warnings %}
{% for w in prompt_result.warnings %}
<p class="muted small"><strong>Uwaga:</strong> {{ w }}</p>
{% endfor %}
{% endif %}
{% if prompt_result.token_plan %}
{% set tp = prompt_result.token_plan %}
<p class="muted small">
Tokeny: prompt {{ tp.prompt_tokens }} · okno modelu {{ tp.context_window }} ·
zarezerwowane na odpowiedź {{ tp.max_output }}
{% if prompt_result.usage and prompt_result.usage.turns and prompt_result.usage.turns > 1 %}
· odpowiedź złożona z {{ prompt_result.usage.turns }} tur (model dokańczał urwany tekst)
{% endif %}
</p>
{% endif %}
{% if prompt_result.llm_error %}
<div class="error">
Nie udało się napisać horoskopu: {{ prompt_result.llm_error }}<br>
Prompt poniżej jest gotowy — możesz go skopiować i użyć ręcznie.
</div>
{% endif %}
{# --- prompt: zawsze dostępny do podglądu i skopiowania --- #}
<div class="meta">
Prompt · {{ st.chars }} znaków (~{{ st.est_tokens }} tokenów) ·
budżet: {{ st.budget }} ·
wskazań: <strong>{{ st.included }}</strong>
{% if st.omitted %}· pominięto: <strong>{{ st.omitted }}</strong>{% endif %}
{% if st.deduplicated %}· scalono powtórek: {{ st.deduplicated }}{% endif %}
</div>
{% if st.omitted %}
<p class="muted small">
Pominięto {{ st.omitted }} najsłabszych wskazań (próg wagi {{ st.min_score_included }}).
Chcesz komplet — wybierz obszerniejszy budżet i wygeneruj ponownie.
</p>
{% endif %}
{% if prompt_result.data_error %}
<div class="error">{{ prompt_result.data_error }} — prompt złożony z samych wyliczeń.</div>
{% endif %}
<div class="actions">
<button type="button" class="ghost" data-copy="#promptText">Kopiuj prompt</button>
<span class="muted small">Możesz też wkleić go samodzielnie do ChatGPT lub Claude.</span>
</div>
<textarea id="promptText" class="prompt" rows="14" readonly>{{ prompt_result.prompt }}</textarea>
{% endif %}
@@ -32,16 +32,6 @@
<option value="porphyry" {{ 'selected' if hs == 'porphyry' else '' }}>Porphyry</option>
</select>
</label>
<label>Zodiak
<select name="zodiac">
{% set zd = form.zodiac or 'tropical' %}
<option value="tropical" {{ 'selected' if zd == 'tropical' else '' }}>Tropikalny</option>
<option value="sidereal_lahiri" {{ 'selected' if zd == 'sidereal_lahiri' else '' }}>Syderyczny (Lahiri)</option>
<option value="sidereal_fagan_bradley" {{ 'selected' if zd == 'sidereal_fagan_bradley' else '' }}>Syderyczny (Fagan-Bradley)</option>
<option value="sidereal_krishnamurti" {{ 'selected' if zd == 'sidereal_krishnamurti' else '' }}>Syderyczny (Krishnamurti)</option>
<option value="draconic" {{ 'selected' if zd == 'draconic' else '' }}>Draconic</option>
</select>
</label>
</div>
{% include "_location_picker.html" %}
<div class="opts">
@@ -65,7 +55,6 @@
Silnik: <strong>{{ result.engine }}</strong> ·
obiektów: {{ result.positions | length }}
{% if result.house_system %}· domy: {{ result.house_system }}{% endif %}
{% if result.zodiac %}· zodiak: {{ result.zodiac }}{% if result.ayanamsha is defined %} (ayanamsa {{ '%.4f'|format(result.ayanamsha) }}°){% endif %}{% endif %}
{% if moment %}· moment: {{ moment }}{% endif %}
</div>
@@ -101,18 +90,6 @@
</tbody>
</table>
{% if result.lots %}
<div class="meta">Lots hermetyczne ({{ result.lots | length }}) · sekta: <strong>{{ result.sect }}</strong></div>
<table class="angles">
<thead><tr><th>Lot</th><th>Znak</th><th>W znaku</th><th>Dom</th><th>Formuła</th></tr></thead>
<tbody>
{% for l in result.lots %}
<tr><td>{{ l.name }}</td><td>{{ l.sign }}</td><td class="mono">{{ l.in_sign }}</td><td>{{ l.house }}</td><td class="muted small">{{ l.formula }}</td></tr>
{% endfor %}
</tbody>
</table>
{% endif %}
{% set with_stations = result.positions | selectattr('stations', 'defined') | list %}
{% if with_stations %}
<div class="meta">Stacje planet (poprzednia / następna; <span class="badge">blisko</span> = mniej niż 7 dni)</div>
@@ -28,12 +28,9 @@
<label><input type="checkbox" name="group" value="true" {{ 'checked' if form.group else '' }}> grupuj identyczne opisy</label>
</div>
{% if location_label %}<p class="muted small">Wstępnie wpisano lokalizację: <strong>{{ location_label }}</strong> ({{ form.lat }}, {{ form.lon }}). Zmień pola lub kliknij „Tu i teraz".</p>{% endif %}
{% include "_prompt_block.html" %}
<div class="actions">
<button type="button" id="nowBtn" class="ghost">Tu i teraz</button>
<button type="submit" name="action" value="report">Szukaj interpretacji</button>
<button type="submit" name="action" value="prompt" class="ghost">Generuj prompt (AI)</button>
<button type="submit" name="action" value="horoscope">Napisz horoskop (AI)</button>
<button type="submit">Szukaj interpretacji</button>
<span id="geoNote" class="muted small"></span>
</div>
</form>
@@ -89,6 +86,4 @@
{% endif %}
<script src="/static/now.js"></script>
<script src="/static/copy.js"></script>
<script src="/static/models.js"></script>
{% endblock %}
@@ -33,12 +33,9 @@
<input type="date" name="to_date" value="{{ form.to_date or '' }}" required>
</label>
</div>
{% include "_prompt_block.html" %}
<div class="actions">
<button type="button" id="nowBtn" class="ghost">Tu i teraz</button>
<button type="submit" name="action" value="timeline">Pokaż kalendarz</button>
<button type="submit" name="action" value="prompt" class="ghost">Generuj prompt (AI)</button>
<button type="submit" name="action" value="horoscope">Napisz horoskop (AI)</button>
<button type="submit">Pokaż kalendarz</button>
<span id="geoNote" class="muted small"></span>
</div>
</form>
@@ -79,6 +76,4 @@
{% endif %}
<script src="/static/now.js"></script>
<script src="/static/copy.js"></script>
<script src="/static/models.js"></script>
{% endblock %}
@@ -1,2 +0,0 @@
-r requirements.txt
pytest>=8.0
-2
View File
@@ -3,5 +3,3 @@ 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
@@ -1,155 +0,0 @@
"""Niezmiennik: KAŻDE wyjście HTTP w dół niesie token międzywarstwowy (LOG-32)
oraz klucz szyfrujący łącze (PRE-16).
Powód istnienia tego testu: token dodano do klienta na gałęzi, która odbiła się od
mastera zanim powstały metody `prompt()` i `horoscope()`. Git zmergował obie zmiany
czysto (różne linie), ale nowe metody wyszły BEZ tokenu — i dostawały 401 dopiero na
produkcji. Zwykły test jednej metody by tego nie złapał, więc sprawdzamy regułę
strukturalnie.
Po dołożeniu szyfrowania ta sama klasa błędu ma gorszy objaw: wywołanie bez `link=`
nie wywala się widocznie, tylko po cichu wysyła treść JAWNYM tekstem. Dlatego dla
wywołań przez `link_crypto` wymagamy obu argumentów naraz.
"""
import ast
import pathlib
CLIENT = pathlib.Path(__file__).resolve().parents[1] / "app" / "clients" / "logic_client.py"
HTTP_VERBS = ("post", "get", "put", "patch", "delete")
LINK_CALLS = ("call", "call_json")
def _http_calls(path: pathlib.Path) -> list[tuple[str, int, bool, bool]]:
"""(opis, linia, czy_ma_headers, czy_wymaga_i_ma_link) dla każdego wyjścia w dół."""
tree = ast.parse(path.read_text(encoding="utf-8"))
out = []
for node in ast.walk(tree):
if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Attribute):
continue
target = node.func.value
has_headers = any(kw.arg == "headers" for kw in node.keywords)
has_link = any(kw.arg == "link" for kw in node.keywords)
if (node.func.attr in HTTP_VERBS
and isinstance(target, ast.Name) and target.id == "client"):
# Surowe wywołanie httpx — omija szyfrowanie, więc dopuszczalne tylko
# dla ścieżek wyjętych spod niego (patrz link_crypto.PUBLIC_PATHS).
out.append((f"client.{node.func.attr}()", node.lineno, has_headers, True))
elif (node.func.attr in LINK_CALLS
and isinstance(target, ast.Name) and target.id == "link_crypto"):
out.append((f"link_crypto.{node.func.attr}()", node.lineno, has_headers, has_link))
return out
def test_client_module_exists():
assert CLIENT.is_file()
def test_every_outbound_call_sends_auth_header():
calls = _http_calls(CLIENT)
assert calls, "nie znaleziono żadnego wywołania HTTP — test przestał cokolwiek pilnować"
missing = [f"{CLIENT.name}:{line} {what}" for what, line, ok, _ in calls if not ok]
assert not missing, (
"Wywołania w dół bez tokenu międzywarstwowego (dostaną 401 przy włączonej "
"ochronie): " + ", ".join(missing)
)
def test_every_outbound_call_passes_link_key():
"""Brak `link=` nie boli od razu — po prostu treść leci jawnym tekstem."""
calls = _http_calls(CLIENT)
missing = [f"{CLIENT.name}:{line} {what}" for what, line, _, ok in calls if not ok]
assert not missing, (
"Wywołania w dół bez klucza łącza — poszłyby NIEZASZYFROWANE: " + ", ".join(missing)
)
def test_raw_http_calls_only_on_paths_exempt_from_encryption():
"""Surowe `client.get/post` wolno wołać wyłącznie tam, gdzie szyfrowania nie ma
z założenia (`/health` dla sond k8s). Każde inne to obejście PRE-16."""
import re
from app import link_crypto
source = CLIENT.read_text(encoding="utf-8").splitlines()
offenders = []
for what, line, _, _ in _http_calls(CLIENT):
if not what.startswith("client."):
continue
url = re.search(r'f"\{self\.base_url\}([^"]*)"', source[line - 1])
if url is None or url.group(1) not in link_crypto.PUBLIC_PATHS:
offenders.append(f"{CLIENT.name}:{line} {what}")
assert not offenders, (
"Surowe wywołania HTTP poza ścieżkami wyjętymi spod szyfrowania: "
+ ", ".join(offenders)
)
def test_auth_headers_helper_is_lazy():
"""Token czytany przy wywołaniu, nie przy imporcie — inaczej pod wystartowałby
z pustym tokenem, gdyby zmienna pojawiła się później."""
import os
from app.clients.logic_client import _auth_headers
old = os.environ.get("INTERNAL_TOKEN")
try:
os.environ["INTERNAL_TOKEN"] = "abc"
assert _auth_headers() == {"X-Astrololo-Token": "abc"}
os.environ.pop("INTERNAL_TOKEN")
assert _auth_headers() == {} # ochrona wyłączona = brak nagłówka
finally:
if old is not None:
os.environ["INTERNAL_TOKEN"] = old
else:
os.environ.pop("INTERNAL_TOKEN", None)
# --------------------------------------------- wybor modelu musi dojsc do logiki
# Ta sama klasa bledu co przy tokenie: dokladajac nowa sciezke latwo zapomniec
# przekazac parametr, a objaw (cichy powrot do modelu domyslnego) jest niewidoczny.
def _calls_to(path_fragment: str) -> list[int]:
"""Linie wywolan logic.<metoda>(...) w handlerach prezentacji."""
main = CLIENT.parent.parent / "main.py"
tree = ast.parse(main.read_text(encoding="utf-8"))
out = []
for node in ast.walk(tree):
if (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
and node.func.attr == path_fragment
and isinstance(node.func.value, ast.Name) and node.func.value.id == "logic"):
out.append(node.lineno)
return out
def _has_kwarg(main_src: str, lineno: int, name: str) -> bool:
tree = ast.parse(main_src)
for node in ast.walk(tree):
if isinstance(node, ast.Call) and node.lineno == lineno:
return any(kw.arg == name for kw in node.keywords)
return False
def test_every_llm_call_passes_selected_model():
main = CLIENT.parent.parent / "main.py"
src = main.read_text(encoding="utf-8")
missing = []
for method in ("prompt", "horoscope"):
for line in _calls_to(method):
if not _has_kwarg(src, line, "model"):
missing.append(f"main.py:{line} logic.{method}()")
assert not missing, (
"Wywołania bez wybranego modelu — po cichu użyją domyślnego: " + ", ".join(missing)
)
def test_every_llm_call_passes_provider():
main = CLIENT.parent.parent / "main.py"
src = main.read_text(encoding="utf-8")
missing = []
for method in ("prompt", "horoscope"):
for line in _calls_to(method):
if not _has_kwarg(src, line, "provider"):
missing.append(f"main.py:{line} logic.{method}()")
assert not missing, "Wywołania bez dostawcy: " + ", ".join(missing)
@@ -1,201 +0,0 @@
"""Logowanie do aplikacji i limit żądań (LOG-32).
To jest brama chroniąca oryginalne bazy interpretacyjne — bez niej każdy w sieci
mógł je wypompować przez `/significators` czy generator promptu. Testy pilnują,
że brama faktycznie zamyka, a nie tylko wygląda na zamkniętą.
"""
import base64
import pytest
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from starlette.testclient import TestClient
from app import security
USER, PASSWORD = "astrololo", "haslo-testowe"
def _basic(user: str, password: str) -> dict[str, str]:
raw = base64.b64encode(f"{user}:{password}".encode()).decode()
return {"Authorization": f"Basic {raw}"}
def _app() -> FastAPI:
app = FastAPI()
security.install(app)
@app.get("/health")
def health():
return {"status": "ok"}
@app.get("/significators", response_class=HTMLResponse)
def significators():
return "<p>treść z bazy interpretacyjnej</p>"
return app
@pytest.fixture(autouse=True)
def _reset_rate_limit():
security._hits.clear()
yield
security._hits.clear()
@pytest.fixture
def guarded(monkeypatch):
monkeypatch.setenv("APP_PASSWORD", PASSWORD)
monkeypatch.setenv("RATE_LIMIT_PER_MIN", "120")
return TestClient(_app())
# ------------------------------------------------------------------ logowanie
def test_blocks_anonymous_access(guarded):
r = guarded.get("/significators")
assert r.status_code == 401
assert "treść z bazy" not in r.text
def test_challenges_with_basic_realm(guarded):
"""Bez nagłówka WWW-Authenticate przeglądarka nie pokaże okna logowania."""
assert "Basic" in guarded.get("/significators").headers.get("WWW-Authenticate", "")
def test_rejects_wrong_password(guarded):
assert guarded.get("/significators", headers=_basic(USER, "zle")).status_code == 401
def test_rejects_wrong_user(guarded):
assert guarded.get("/significators", headers=_basic("obcy", PASSWORD)).status_code == 401
def test_rejects_malformed_header(guarded):
for bad in ("Basic !!!niebase64!!!", "Bearer cokolwiek", "", "Basic"):
assert guarded.get("/significators", headers={"Authorization": bad}).status_code == 401
def test_allows_correct_credentials(guarded):
r = guarded.get("/significators", headers=_basic(USER, PASSWORD))
assert r.status_code == 200
assert "treść z bazy" in r.text
def test_health_stays_public(guarded):
assert guarded.get("/health").status_code == 200
def test_open_when_password_unset(monkeypatch):
"""Brak hasła = zgodność wstecz; ochrona wyłączona (i ostrzegamy przy starcie)."""
monkeypatch.delenv("APP_PASSWORD", raising=False)
assert not security.auth_enabled()
assert TestClient(_app()).get("/significators").status_code == 200
# --------------------------------------------------------------- limit żądań
def test_rate_limit_blocks_flood(monkeypatch):
"""Masowe odpytywanie to droga eksfiltracji nawet po zalogowaniu."""
monkeypatch.setenv("APP_PASSWORD", PASSWORD)
monkeypatch.setenv("RATE_LIMIT_PER_MIN", "5")
client = TestClient(_app())
auth = _basic(USER, PASSWORD)
codes = [client.get("/significators", headers=auth).status_code for _ in range(8)]
assert codes[:5] == [200] * 5
assert 429 in codes[5:]
def test_rate_limited_response_has_retry_after(monkeypatch):
monkeypatch.setenv("APP_PASSWORD", PASSWORD)
monkeypatch.setenv("RATE_LIMIT_PER_MIN", "1")
client = TestClient(_app())
auth = _basic(USER, PASSWORD)
client.get("/significators", headers=auth)
r = client.get("/significators", headers=auth)
assert r.status_code == 429 and r.headers.get("Retry-After") == "60"
def test_rate_limit_precedes_auth(monkeypatch):
"""Limit musi działać także dla niezalogowanych — inaczej zgadywanie hasła
i sondowanie API jest darmowe."""
monkeypatch.setenv("APP_PASSWORD", PASSWORD)
monkeypatch.setenv("RATE_LIMIT_PER_MIN", "3")
client = TestClient(_app())
codes = [client.get("/significators").status_code for _ in range(6)]
assert 429 in codes
def test_rate_limit_disabled_when_zero(monkeypatch):
monkeypatch.setenv("RATE_LIMIT_PER_MIN", "0")
monkeypatch.delenv("APP_PASSWORD", raising=False)
client = TestClient(_app())
assert all(client.get("/significators").status_code == 200 for _ in range(30))
# ------------------------------------------ adres klienta za odwrotnym proxy
#
# Po włączeniu TLS (PRE-16) aplikacja stoi za Ingressem, więc bezpośredni peer
# to zawsze POD proxy. Te testy pilnują obu stron kompromisu: żeby limit dalej
# rozróżniał ludzi, a jednocześnie żeby nagłówek nie stał się furtką do jego
# ominięcia.
class _Req:
"""Minimalny zamiennik Request — `client_ip` czyta tylko te dwa pola."""
def __init__(self, peer: str | None, **headers: str):
self.client = type("C", (), {"host": peer})() if peer else None
self.headers = {k.replace("_", "-"): v for k, v in headers.items()}
def test_client_ip_ignores_headers_without_trust_proxy(monkeypatch):
"""Bez TRUST_PROXY nagłówek jest bezwartościowy — każdy może go dopisać."""
monkeypatch.delenv("TRUST_PROXY", raising=False)
req = _Req("10.42.0.7", x_forwarded_for="1.2.3.4", x_real_ip="5.6.7.8")
assert security.client_ip(req) == "10.42.0.7"
def test_client_ip_takes_last_forwarded_entry(monkeypatch):
"""Ostatni wpis dopisał NASZ proxy; wcześniejsze mógł podstawić klient."""
monkeypatch.setenv("TRUST_PROXY", "true")
req = _Req("10.42.0.7", x_forwarded_for="1.1.1.1, 2.2.2.2, 192.168.1.50")
assert security.client_ip(req) == "192.168.1.50"
def test_client_ip_falls_back_to_real_ip(monkeypatch):
monkeypatch.setenv("TRUST_PROXY", "true")
req = _Req("10.42.0.7", x_real_ip="192.168.1.50")
assert security.client_ip(req) == "192.168.1.50"
def test_client_ip_falls_back_to_peer_when_headers_missing(monkeypatch):
monkeypatch.setenv("TRUST_PROXY", "true")
assert security.client_ip(_Req("10.42.0.7")) == "10.42.0.7"
def test_spoofed_forwarded_header_cannot_dodge_the_limit(monkeypatch):
"""Sedno sprawy: bez zaufania do proxy podszywanie się NIE resetuje wiadra.
Gdyby limiter brał pierwszy lepszy `X-Forwarded-For`, wystarczyłoby zmieniać
go co żądanie, żeby pompować bazę bez ograniczeń.
"""
monkeypatch.delenv("TRUST_PROXY", raising=False)
monkeypatch.setenv("RATE_LIMIT_PER_MIN", "3")
monkeypatch.delenv("APP_PASSWORD", raising=False)
client = TestClient(_app())
codes = [client.get("/significators", headers={"X-Forwarded-For": f"9.9.9.{i}"}).status_code
for i in range(6)]
assert 429 in codes
def test_proxied_clients_get_separate_buckets(monkeypatch):
"""Za proxy dwie różne osoby nie mogą się nawzajem odcinać."""
monkeypatch.setenv("TRUST_PROXY", "true")
monkeypatch.setenv("RATE_LIMIT_PER_MIN", "3")
monkeypatch.delenv("APP_PASSWORD", raising=False)
client = TestClient(_app())
first = [client.get("/significators", headers={"X-Forwarded-For": "192.168.1.50"}).status_code
for _ in range(5)]
second = client.get("/significators", headers={"X-Forwarded-For": "192.168.1.51"})
assert 429 in first, "limit musi zadziałać dla pierwszego adresu"
assert second.status_code == 200, "drugi adres ma własne wiadro"