feat(render): raport PDF przez LaTeX jako osobna usluga (PRE-24)
Testy / Testy warstwy logicznej (silnik) (push) Successful in 11m4s
Testy / Testy warstwy prezentacji (dostęp do baz) (push) Successful in 9m57s
Testy / Build obrazu silnika B (swisseph) (push) Successful in 34s
Testy / Kontrola składni wszystkich warstw (push) Successful in 22s

Ostatnia z osmiu wskazowek partnerow. Nowa usluga services/render sklada raport:
generuje plik posredni .tex i kompiluje go XeLaTeX-em do PDF.

OSOBNY komponent, nie czesc prezentacji — ta sama zasada co przy izolacji
swissepha (LOG-27). TeX Live wazy setki megabajtow; w obrazie produktu
spowalnialby kazdy build, a tak aktualizuje sie niezaleznie i jego awaria nie
kladzie aplikacji, tylko przycisk „Pobierz PDF".

SZYFROWANIE, o ktore prosiles: lacze prezentacja↔render idzie tak samo jak
pozostale, ale z WLASNYM, TRZECIM kluczem (LINK_KEY_PRESENTATION_RENDER). Osobny,
bo tym laczem plynie CALY raport — dane urodzeniowe i opisy z baz — wiec przejecie
go nie moze otwierac lacza do logiki ani do danych. Fail-closed: bez klucza pod
nie wstaje. Test kopii link_crypto obejmuje teraz cztery uslugi.

Uklad PDF wg prosby: imie i nazwisko -> wprowadzone dane -> RYSUNEK kosmogramu
-> interpretacja natalna -> predykcje okresowe. Test pilnuje kolejnosci.

Dwie rzeczy, ktore wyszly dopiero przy skladaniu tego kawalka:

1. KOSMOGRAM NIE NADAWAL SIE DO PDF. Uzywa zmiennych CSS (var(--line)) i klasy
   .glyph, ktorej font podaje styles.css — a samodzielny konwerter SVG→PDF nie zna
   naszego arkusza. Wyszlyby czarne kreski BEZ SYMBOLI. Stad wariant „print":
   konkretne kolory na bialym tle i font glifow wpisany wprost w rysunek. Przy
   okazji zalatwia to etap 4 planu kola (wariant do druku).

2. UCIECZKA ZNAKOW LATEXA byla zepsuta — i zlapal to moj wlasny test. Zamiana
   „\” na \textbackslash{} szla w tej samej petli co nawiasy, wiec kolejne
   podmiany ucieklyby nawiasy dopiero co wstawione: wychodzilo
   \textbackslash\{\}. Poprawka: ukosnik chowany pod znacznik zastepczy i
   rozwijany na koncu. To nie kosmetyka — niezauwazony „%” komentuje RESZTE LINII,
   wiec zdanie od modelu urywaloby sie w polowie, a PDF powstawalby normalnie,
   tylko krotszy. Test na wrogim tekscie sprawdza tez, ze \end{document} ani
   \input{} nie wyrwa sie z dokumentu.

Usluga nie zapisuje nic poza katalogiem tymczasowym, ktory sprzata po sobie;
w manifescie readOnlyRootFilesystem + emptyDir na /tmp. /health raportuje
obecnosc xelatex i rsvg-convert, zeby zepsuty obraz bylo widac od razu.

Zweryfikowane na zywo (TestClient uslugi render): zadanie bez szyfrowania
z POPRAWNYM tokenem -> 400; obcy klucz -> 400 i tajny opis nie wraca; wlasciwy
klucz -> zadanie dochodzi do aplikacji, tresci baz NIE MA na kablu, odpowiedz
zaszyfrowana. Manifesty przechodza kubectl apply --dry-run=server na zywym
klastrze.

CZEGO NIE SPRAWDZILEM: samej kompilacji PDF. W tym srodowisku nie ma ani TeX
Live, ani dzialajacego runtime'u kontenerow (docker CLI jest, daemon nie) —
probowalem zbudowac obraz i sie nie dalo. Sprawdzone jest wszystko dookola:
generowanie .tex, ucieczka znakow, szyfrowanie, kontrakt API, samowystarczalnosc
SVG. Pierwsze uruchomienie na klastrze trzeba obejrzec — dlatego PRE-24 zostaje
jako „W trakcie", nie „Zrobione".

Instrukcja wdrozenia: docs/wdrozenie-render-pdf.md (klucz, obraz, merge,
weryfikacja, znane ograniczenia).

Testy: 15 nowych (render) + 14 (lacze i wariant druku w prezentacji).
Calosc: prezentacja 131, logika 265/1 skip, render 15.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-24 19:21:49 +02:00
parent 6c9029c496
commit f24616d342
22 changed files with 1546 additions and 22 deletions
+4
View File
@@ -65,6 +65,10 @@ CONTENT_TYPE = "application/vnd.astrololo.enc"
ENV_PRESENTATION_LOGIC = "LINK_KEY_PRESENTATION_LOGIC"
ENV_LOGIC_DATA = "LINK_KEY_LOGIC_DATA"
# Trzecia para: prezentacja ↔ render (PRE-24). Osobny klucz, jak przy pozostałych —
# usługa render dostaje CAŁY raport (dane urodzeniowe + opisy z baz), więc przejęcie
# jej klucza nie może otwierać łącza do logiki ani do danych.
ENV_PRESENTATION_RENDER = "LINK_KEY_PRESENTATION_RENDER"
ENV_REQUIRED = "LINK_ENCRYPTION_REQUIRED"
REQUEST, RESPONSE = b"req", b"res"
+4
View File
@@ -65,6 +65,10 @@ CONTENT_TYPE = "application/vnd.astrololo.enc"
ENV_PRESENTATION_LOGIC = "LINK_KEY_PRESENTATION_LOGIC"
ENV_LOGIC_DATA = "LINK_KEY_LOGIC_DATA"
# Trzecia para: prezentacja ↔ render (PRE-24). Osobny klucz, jak przy pozostałych —
# usługa render dostaje CAŁY raport (dane urodzeniowe + opisy z baz), więc przejęcie
# jej klucza nie może otwierać łącza do logiki ani do danych.
ENV_PRESENTATION_RENDER = "LINK_KEY_PRESENTATION_RENDER"
ENV_REQUIRED = "LINK_ENCRYPTION_REQUIRED"
REQUEST, RESPONSE = b"req", b"res"
+3 -3
View File
@@ -342,13 +342,13 @@ def test_incremental_unframing_handles_split_frames(link):
# ----------------------------------------------- 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
def test_all_services_share_the_same_module():
"""Moduł jest skopiowany do KAŻDEJ usługi, która szyfruje łącze (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")}
for svc in ("presentation", "logic", "data", "render")}
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()}
+53 -19
View File
@@ -49,6 +49,22 @@ MIN_SEP = 8.0
# muted kolory żywiołów — czytelne na ciemnym tle, spójne z paletą aplikacji
_ELEMENT_COLOR = ["#d98a6a", "#8fae7a", "#cfc06a", "#6aa6c9"] # ogień, ziemia, powietrze, woda
# ── motywy ───────────────────────────────────────────────────────────────
# „screen" korzysta ze zmiennych CSS strony, więc koło samo dostraja się do
# motywu aplikacji. „print" MUSI mieć konkretne kolory: SVG idzie wtedy do
# osobnego konwertera (rsvg → PDF, PRE-24), który o naszym arkuszu nic nie wie
# i zmiennych `var(--…)` nie rozwiąże — wyszłyby czarne albo żadne kreski.
# Przy okazji odwracamy kontrast: raport drukuje się na białym papierze.
_THEMES = {
"screen": {"line": "var(--line)", "ink": "var(--ink)", "muted": "var(--muted)",
"accent": "var(--accent)", "disc": "#12132a", "retro": "#ff9b6a",
"elements": _ELEMENT_COLOR},
"print": {"line": "#b9bcc9", "ink": "#111318", "muted": "#5d6270",
"accent": "#5a49c8", "disc": "#ffffff", "retro": "#b2452a",
# ciemniejsze warianty żywiołów — na białym tle jasne byłyby niewidoczne
"elements": ["#a8442a", "#3f6b33", "#8a6d10", "#1f5c80"]},
}
def _pt(lon: float, asc: float, r: float) -> tuple[float, float]:
phi = math.radians(180.0 - (lon - asc))
@@ -61,10 +77,20 @@ def _line(x1: float, y1: float, x2: float, y2: float, stroke: str, w: float = 1.
f'stroke="{stroke}" stroke-width="{w}" opacity="{opacity}"/>')
def _text(x: float, y: float, s: str, *, size: float, fill: str, cls: str = "") -> str:
# Stack fontów z glifami astrologicznymi. Na stronie dostarcza go klasa CSS
# `.glyph`, ale w wariancie do druku SVG jedzie do SAMODZIELNEGO konwertera, który
# naszego arkusza nie widzi — wtedy trzeba wpisać font wprost do elementu, inaczej
# rysunek wychodzi BEZ symboli (konwerter bierze domyślny font, a ten ich nie ma).
GLYPH_FONT = ("Apple Symbols, Segoe UI Symbol, Noto Sans Symbols2, "
"DejaVu Sans, sans-serif")
def _text(x: float, y: float, s: str, *, size: float, fill: str, cls: str = "",
font: str = "") -> str:
c = f' class="{cls}"' if cls else ""
f = f' font-family="{font}"' if font else ""
return (f'<text x="{x:.2f}" y="{y:.2f}" font-size="{size}" fill="{fill}"'
f' text-anchor="middle" dominant-baseline="central"{c}>{escape(s)}</text>')
f' text-anchor="middle" dominant-baseline="central"{c}{f}>{escape(s)}</text>')
def spread(lons: list[float], min_sep: float = MIN_SEP, passes: int = 300) -> list[float]:
@@ -126,8 +152,16 @@ def _delta(a: float, b: float) -> float:
return abs(((a - b + 180.0) % 360.0) - 180.0)
def render(chart: dict) -> str:
"""Zwraca SVG koła albo '' gdy brakuje danych (silnik bez osi/domów)."""
def render(chart: dict, theme: str = "screen") -> str:
"""Zwraca SVG koła albo '' gdy brakuje danych (silnik bez osi/domów).
theme='screen' — kolory ze zmiennych CSS aplikacji (dostraja się do motywu).
theme='print' — konkretne kolory na białym tle; konieczne, gdy SVG trafia
do samodzielnego konwertera (PDF), który zmiennych CSS nie rozwiąże.
"""
T = _THEMES.get(theme, _THEMES["screen"])
# tylko dla druku — na stronie font podaje CSS (.glyph)
glyph_font = GLYPH_FONT if theme == "print" else ""
angles = chart.get("angles")
cusps = chart.get("cusps")
signs = chart.get("sign_glyphs")
@@ -138,26 +172,26 @@ def render(chart: dict) -> str:
parts: list[str] = []
# tło i okręgi
parts.append(f'<circle cx="{_CX}" cy="{_CY}" r="{R_OUT}" fill="#12132a" stroke="var(--line)" stroke-width="1.5"/>')
parts.append(f'<circle cx="{_CX}" cy="{_CY}" r="{R_ZOD}" fill="none" stroke="var(--line)" stroke-width="1"/>')
parts.append(f'<circle cx="{_CX}" cy="{_CY}" r="{R_HUB}" fill="none" stroke="var(--line)" stroke-width="1"/>')
parts.append(f'<circle cx="{_CX}" cy="{_CY}" r="{R_OUT}" fill="{T["disc"]}" stroke="{T["line"]}" stroke-width="1.5"/>')
parts.append(f'<circle cx="{_CX}" cy="{_CY}" r="{R_ZOD}" fill="none" stroke="{T["line"]}" stroke-width="1"/>')
parts.append(f'<circle cx="{_CX}" cy="{_CY}" r="{R_HUB}" fill="none" stroke="{T["line"]}" stroke-width="1"/>')
# drobne podziałki co 5°, mocniejsze co 10° (na pasie znaków)
for deg in range(0, 360, 5):
r_in = R_TICK if deg % 10 else R_TICK - 4
x1, y1 = _pt(deg, asc, R_ZOD)
x2, y2 = _pt(deg, asc, r_in)
parts.append(_line(x1, y1, x2, y2, "var(--line)", 0.6, 0.7))
parts.append(_line(x1, y1, x2, y2, T["line"], 0.6, 0.7))
# pas znaków: granice co 30° + glif znaku w środku sektora, kolorem żywiołu
for i, sg in enumerate(signs):
b = 30.0 * i
x1, y1 = _pt(b, asc, R_ZOD)
x2, y2 = _pt(b, asc, R_OUT)
parts.append(_line(x1, y1, x2, y2, "var(--line)", 1.0))
parts.append(_line(x1, y1, x2, y2, T["line"], 1.0))
gx, gy = _pt(b + 15.0, asc, R_SIGN)
parts.append(_text(gx, gy, sg.get("glyph") or "", size=17,
fill=_ELEMENT_COLOR[i % 4], cls="glyph"))
fill=T["elements"][i % 4], cls="glyph", font=glyph_font))
# szprychy domów (od pasa do piasty) + numery domów w środku każdego domu
n = len(cusps)
@@ -165,24 +199,24 @@ def render(chart: dict) -> str:
lon = float(c["decimal"])
x1, y1 = _pt(lon, asc, R_ZOD)
x2, y2 = _pt(lon, asc, R_HUB)
parts.append(_line(x1, y1, x2, y2, "var(--line)", 0.8, 0.85))
parts.append(_line(x1, y1, x2, y2, T["line"], 0.8, 0.85))
nxt = float(cusps[(i + 1) % n]["decimal"])
span = (nxt - lon) % 360.0 or 360.0
mid = lon + span / 2.0
hx, hy = _pt(mid, asc, R_HNUM)
parts.append(_text(hx, hy, str(c["house"]), size=11, fill="var(--muted)"))
parts.append(_text(hx, hy, str(c["house"]), size=11, fill=T["muted"]))
# osie: AscDsc i MCIC przez całe koło, wyróżnione akcentem
for a, b, la, lb in (("Asc", "Dsc", "AC", "DC"), ("MC", "IC", "MC", "IC")):
lon_a = float(angles[a]["decimal"])
ax, ay = _pt(lon_a, asc, R_ZOD)
bx, by = _pt(lon_a + 180.0, asc, R_ZOD)
parts.append(_line(ax, ay, bx, by, "var(--accent)", 1.6, 0.9))
parts.append(_line(ax, ay, bx, by, T["accent"], 1.6, 0.9))
# etykiety POZA kołem — w środku kolidowały z glifami obiektów
lax, lay = _pt(lon_a, asc, R_AXIS_LABEL)
lbx, lby = _pt(lon_a + 180.0, asc, R_AXIS_LABEL)
parts.append(_text(lax, lay, la, size=10, fill="var(--accent)"))
parts.append(_text(lbx, lby, lb, size=10, fill="var(--accent)"))
parts.append(_text(lax, lay, la, size=10, fill=T["accent"]))
parts.append(_text(lbx, lby, lb, size=10, fill=T["accent"]))
# ── obiekty: glify na pozycjach, stopień w znaku, retrogradacja ─────
# Glify rozsuwamy, ale kreska na pasie zostaje w PRAWDZIWEJ pozycji —
@@ -197,15 +231,15 @@ def render(chart: dict) -> str:
mx1, my1 = _pt(true_lon, asc, R_ZOD)
mx2, my2 = _pt(true_lon, asc, R_MARK)
parts.append(_line(mx1, my1, mx2, my2, "var(--ink)", 1.0, 0.75))
parts.append(_line(mx1, my1, mx2, my2, T["ink"], 1.0, 0.75))
if _delta(draw_lon, true_lon) > 0.4:
lx, ly = _pt(draw_lon, asc, R_PLANET + 12)
parts.append(_line(mx2, my2, lx, ly, "var(--line)", 0.7, 0.9))
parts.append(_line(mx2, my2, lx, ly, T["line"], 0.7, 0.9))
gx, gy = _pt(draw_lon, asc, R_PLANET)
parts.append(_text(gx, gy, obj.get("glyph") or "", size=16,
fill="var(--ink)", cls="glyph"))
fill=T["ink"], cls="glyph", font=glyph_font))
# Sam stopień w znaku, bez „°" — tak robią programy astrologiczne i tylko
# tak podpisy mieszczą się obok siebie w skupiskach. Retrogradacja: ℞ oraz
@@ -213,7 +247,7 @@ def render(chart: dict) -> str:
dx, dy = _pt(draw_lon, asc, R_DEG)
label = f"{int(true_lon % 30)}" + ("" if retro else "")
parts.append(_text(dx, dy, label, size=8,
fill="#ff9b6a" if retro else "var(--muted)"))
fill=T["retro"] if retro else T["muted"]))
body = "".join(parts)
return (f'<svg class="wheel" viewBox="0 0 {_VIEW} {_VIEW}" role="img" '
@@ -0,0 +1,52 @@
"""Klient HTTP do usługi render (PRE-24).
Jedyny punkt styku prezentacji z usługą składającą PDF. Idzie po szyfrowanym
łączu, tak jak reszta ruchu wewnątrz klastra — z WŁASNYM, trzecim kluczem
(`LINK_KEY_PRESENTATION_RENDER`). Osobny klucz, bo przez to łącze leci CAŁY
raport: dane urodzeniowe i opisy z baz. Przejęcie go nie może otwierać łącza do
logiki ani do danych.
"""
from __future__ import annotations
import os
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↔render. Czytany przy każdym wywołaniu, bo
konfiguracja może się zmienić bez restartu procesu."""
key = link_crypto.key_from_env(link_crypto.ENV_PRESENTATION_RENDER)
return link_crypto.Link(key) if key else None
class RenderClient:
def __init__(self, base_url: str | None = None) -> None:
self.base_url = (base_url or settings.render_url).rstrip("/")
def pdf(self, report: dict) -> bytes:
"""Raport → bajty PDF. Timeout długi: XeLaTeX na obszernym tekście
potrzebuje chwili, a raport bywa na kilkadziesiąt stron."""
with httpx.Client(timeout=max(settings.http_timeout, 180.0)) as client:
return link_crypto.call(client, "POST", f"{self.base_url}/pdf",
payload=report, headers=_auth_headers(),
link=_link())
def health(self) -> dict:
"""/health celowo poza szyfrowaniem — pukają tu sondy k8s, które klucza
nie mają, a nie przechodzi tędy nic z raportu."""
import json
with httpx.Client(timeout=settings.http_timeout) as client:
r = client.get(f"{self.base_url}/health", headers=_auth_headers())
r.raise_for_status()
return json.loads(r.content)
+2
View File
@@ -11,6 +11,8 @@ from dataclasses import dataclass, field
@dataclass
class Settings:
logic_url: str = field(default_factory=lambda: os.getenv("LOGIC_URL", "http://localhost:8001"))
# Usługa składająca PDF (PRE-24) — osobny komponent, bo dźwiga TeX Live.
render_url: str = field(default_factory=lambda: os.getenv("RENDER_URL", "http://localhost:8004"))
http_timeout: float = field(default_factory=lambda: float(os.getenv("HTTP_TIMEOUT", "10")))
+4
View File
@@ -65,6 +65,10 @@ CONTENT_TYPE = "application/vnd.astrololo.enc"
ENV_PRESENTATION_LOGIC = "LINK_KEY_PRESENTATION_LOGIC"
ENV_LOGIC_DATA = "LINK_KEY_LOGIC_DATA"
# Trzecia para: prezentacja ↔ render (PRE-24). Osobny klucz, jak przy pozostałych —
# usługa render dostaje CAŁY raport (dane urodzeniowe + opisy z baz), więc przejęcie
# jej klucza nie może otwierać łącza do logiki ani do danych.
ENV_PRESENTATION_RENDER = "LINK_KEY_PRESENTATION_RENDER"
ENV_REQUIRED = "LINK_ENCRYPTION_REQUIRED"
REQUEST, RESPONSE = b"req", b"res"
+59
View File
@@ -149,6 +149,65 @@ def compile_build(
return templates.TemplateResponse(request, "compile.html", ctx)
@app.post("/compile/pdf")
def compile_pdf(payload: dict):
"""Składa raport PDF (PRE-24) — woła usługę render po szyfrowanym łączu.
Wejście z przeglądarki, bo części od AI (interpretacja natalna i predykcje)
mieszkają w magazynie lokalnym. Kosmogram i tabele liczymy TU, żeby PDF
zawierał dokładnie to, co widać na stronie.
Kosmogram idzie w motywie DRUKU: samodzielny konwerter SVG→PDF nie zna
naszego arkusza, więc zmienne CSS i font glifów muszą być w samym rysunku.
"""
from fastapi.responses import Response
from app.clients.render_client import RenderClient
data = payload.get("data") or {}
try:
iso_utc, label = _build_utc(
str(data.get("date") or ""), str(data.get("time") or ""),
float(data.get("tz_offset") or 0.0),
)
except (ValueError, TypeError) as e:
return JSONResponse({"detail": f"Niepoprawne dane wejściowe: {e}"}, status_code=422)
wheel_svg = ""
try:
chart = logic.positions(
when_utc_iso=iso_utc,
lat=float(data.get("lat") or 0.0), lon=float(data.get("lon") or 0.0),
house_system=str(data.get("house_system") or "whole_sign"),
zodiac=str(data.get("zodiac") or "tropical"),
)
from app import chartwheel
wheel_svg = chartwheel.render(chart, theme="print")
except httpx.HTTPError as e:
# Brak rysunku nie może zablokować raportu — tekst jest ważniejszy.
log_note = _logic_error(e)
chart, wheel_svg = None, ""
data = {**data, "wheel_error": log_note}
report = {
"person": payload.get("person") or "",
"data": {**data, "moment_utc": label},
"wheel_svg": wheel_svg,
"natal": payload.get("natal") or {},
"predictions": payload.get("predictions") or [],
}
try:
pdf = RenderClient().pdf(report)
except httpx.HTTPError as e:
return JSONResponse(
{"detail": f"Usługa render niedostępna albo nie złożyła PDF-a: {e}"},
status_code=502,
)
return Response(content=pdf, media_type="application/pdf",
headers={"Content-Disposition": 'attachment; filename="raport.pdf"'})
# ---------------- Sygnifikatory (wyszukiwarka w bazach) ----------------
@app.get("/significators", response_class=HTMLResponse)
def significators_form(request: Request):
@@ -85,11 +85,75 @@
else document.addEventListener('DOMContentLoaded', fn);
}
// ── PDF (PRE-24) ────────────────────────────────────────────────────
// Części od AI mieszkają w przeglądarce, więc raport składamy TU i wysyłamy
// do serwera, który dokłada kosmogram i woła usługę render.
function formData() {
function v(name) {
var el = document.querySelector('[name=' + name + ']');
return el ? el.value : '';
}
return {
date: v('date'), time: v('time'), tz_offset: v('tz_offset'),
lat: v('lat'), lon: v('lon'),
house_system: v('house_system'), zodiac: v('zodiac'),
place: (document.querySelector('#geoSearch') || {}).value || ''
};
}
function download(blob) {
var url = URL.createObjectURL(blob);
var a = document.createElement('a');
a.href = url;
a.download = 'raport.pdf';
document.body.appendChild(a);
a.click();
a.remove();
// Zwalniamy obiekt dopiero po kliknięciu — wcześniej przeglądarka nie zdąży
// rozpocząć pobierania.
setTimeout(function () { URL.revokeObjectURL(url); }, 4000);
}
function buildPdf(btn, note) {
var person = (document.querySelector('[name=person]') || {}).value || '';
var payload = {
person: person,
data: formData(),
natal: readNatal() || {},
predictions: readPredictions()
};
btn.disabled = true;
note.textContent = 'Składam PDF — LaTeX potrzebuje chwili przy obszernym tekście…';
fetch('/compile/pdf', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(payload)
}).then(function (r) {
if (!r.ok) {
return r.json().catch(function () { return {}; }).then(function (d) {
throw new Error(d.detail || ('HTTP ' + r.status));
});
}
return r.blob();
}).then(function (blob) {
download(blob);
note.textContent = 'Gotowe — PDF pobrany.';
}).catch(function (e) {
note.textContent = 'Nie udało się złożyć PDF-a: ' + e.message;
}).finally(function () {
btn.disabled = false;
});
}
ready(function () {
if (!document.getElementById('readiness')) return; // nie ta zakładka
// Horoskop jest w dokumencie tylko po wysłaniu formularza.
renderReadiness(!!document.getElementById('reportPerson'));
renderNatal();
renderPredictions();
var btn = document.getElementById('pdfBtn');
var note = document.getElementById('pdfNote');
if (btn && note) btn.addEventListener('click', function () { buildPdf(btn, note); });
});
})();
@@ -57,6 +57,9 @@ zapamiętane predykcje okresowe. Dane pobiera z pozostałych zakładek — nie t
</div>
<div class="actions">
<button type="submit">Złóż podsumowanie</button>
<button type="button" id="pdfBtn" class="ghost"
title="Generuje plik pośredni LaTeXa i kompiluje go do PDF (usługa render)">Pobierz PDF</button>
<span id="pdfNote" class="muted small"></span>
</div>
</form>
@@ -0,0 +1,149 @@
"""Łącze do usługi render i wariant kosmogramu do druku (PRE-24).
Przez to łącze leci CAŁY raport: dane urodzeniowe i opisy z baz. Musi być
szyfrowane tak samo jak pozostałe, WŁASNYM kluczem — i musi obowiązywać ten sam
niezmiennik co dla klienta logiki: żadnego wyjścia w dół bez tokenu i klucza.
"""
import pathlib
import pytest
APP = pathlib.Path(__file__).resolve().parents[1] / "app"
CLIENT = APP / "clients" / "render_client.py"
SRC = CLIENT.read_text(encoding="utf-8")
WHEEL = (APP / "chartwheel.py").read_text(encoding="utf-8")
MAIN = (APP / "main.py").read_text(encoding="utf-8")
COMPILE_JS = (APP / "static" / "compile.js").read_text(encoding="utf-8")
# ─────────────────────────── szyfrowanie łącza ───────────────────────────
def test_uses_its_own_third_key():
"""Osobny klucz na parę rozmówców. Przejęcie klucza renderu nie może otwierać
łącza do logiki ani do danych."""
from app import link_crypto
assert link_crypto.ENV_PRESENTATION_RENDER == "LINK_KEY_PRESENTATION_RENDER"
assert link_crypto.ENV_PRESENTATION_RENDER != link_crypto.ENV_PRESENTATION_LOGIC
assert link_crypto.ENV_PRESENTATION_RENDER != link_crypto.ENV_LOGIC_DATA
assert "ENV_PRESENTATION_RENDER" in SRC
def test_report_goes_through_the_encrypted_link():
assert "link_crypto.call(" in SRC
assert "link=_link()" in SRC
def test_every_outbound_call_carries_token_and_key():
"""Ten sam niezmiennik co przy kliencie logiki — tam brak nagłówka wyszedł
dopiero na produkcji (401)."""
import ast
tree = ast.parse(SRC)
calls = [n for n in ast.walk(tree)
if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)
and n.func.attr in ("call", "call_json")]
assert calls, "nie znaleziono wywołania przez link_crypto"
for c in calls:
kwargs = {k.arg for k in c.keywords}
assert "headers" in kwargs and "link" in kwargs
def test_health_is_the_only_unencrypted_path():
"""/health musi zostać otwarty dla sond k8s — ale nic poza nim."""
from app import link_crypto
assert "/health" in link_crypto.PUBLIC_PATHS
raw = [line for line in SRC.splitlines() if "client.get(" in line or "client.post(" in line]
assert all("/health" in line for line in raw), f"surowe wywołania poza /health: {raw}"
def test_key_is_read_lazily():
"""Konfiguracja może się zmienić bez restartu procesu (podmiana sekretu)."""
assert "def _link()" in SRC and "key_from_env" in SRC
# ──────────────────────── kosmogram w wariancie do druku ─────────────────
def test_pdf_route_uses_print_theme():
"""Samodzielny konwerter SVG→PDF nie zna naszego arkusza stylów."""
assert 'chartwheel.render(chart, theme="print")' in MAIN
def test_print_theme_has_no_css_variables():
"""`var(--…)` w SVG poza stroną nie rozwiąże się — wyszłyby czarne albo
żadne kreski."""
from app import chartwheel
chart = _sample_chart()
printed = chartwheel.render(chart, theme="print")
assert printed and "var(--" not in printed
def test_screen_theme_still_uses_css_variables():
"""Na stronie koło ma się dostrajać do motywu aplikacji."""
from app import chartwheel
assert "var(--" in chartwheel.render(_sample_chart(), theme="screen")
def test_print_theme_embeds_the_glyph_font():
"""Font glifów podaje CSS klasy `.glyph` — poza stroną trzeba go wpisać
W SAM rysunek, inaczej kosmogram w PDF wyszedłby BEZ symboli."""
from app import chartwheel
printed = chartwheel.render(_sample_chart(), theme="print")
assert 'font-family=' in printed
assert "DejaVu Sans" in printed # font obecny w obrazie usługi render
def test_screen_theme_leaves_font_to_css():
from app import chartwheel
assert "font-family=" not in chartwheel.render(_sample_chart(), theme="screen")
def test_unknown_theme_falls_back_to_screen():
from app import chartwheel
assert chartwheel.render(_sample_chart(), theme="cokolwiek") == \
chartwheel.render(_sample_chart(), theme="screen")
# ─────────────────────────── składanie po stronie klienta ────────────────
def test_browser_sends_all_three_parts():
for field in ("person", "natal", "predictions"):
assert field in COMPILE_JS, f"raport wysyłany bez części: {field}"
def test_pdf_button_exists():
tpl = (APP / "templates" / "compile.html").read_text(encoding="utf-8")
assert 'id="pdfBtn"' in tpl
def test_wheel_failure_does_not_block_the_report():
"""Brak rysunku ma dać raport bez kosmogramu, a nie brak raportu — tekst
kosztował wywołanie modelu, rysunek policzymy zawsze."""
assert "wheel_svg = \"\"" in MAIN or 'wheel_svg, = ""' in MAIN
assert "Brak rysunku nie może zablokować raportu" in MAIN
# ─────────────────────────────────── pomocnicze ──────────────────────────
def _sample_chart():
signs = ["Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo",
"Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces"]
glyphs = ["", "", "", "", "", "", "", "", "", "", "", ""]
asc = 128.9
asc_sign = int(asc // 30)
return {
"angles": {"Asc": {"name": "Asc", "decimal": asc, "sign": signs[asc_sign]},
"MC": {"name": "MC", "decimal": 5.0},
"Dsc": {"name": "Dsc", "decimal": (asc + 180) % 360},
"IC": {"name": "IC", "decimal": 185.0}},
"cusps": [{"house": i + 1, "sign": signs[(asc_sign + i) % 12],
"decimal": (asc_sign * 30 + i * 30) % 360} for i in range(12)],
"sign_glyphs": [{"sign": s, "glyph": g} for s, g in zip(signs, glyphs)],
"positions": [{"name": "Sun", "decimal": 40.2, "glyph": "", "direction": "D"}],
}
+31
View File
@@ -0,0 +1,31 @@
# Usługa render: składa raport PDF (PRE-24).
#
# Obraz jest CIĘŻKI (TeX Live) i właśnie dlatego stoi osobno — patrz app/main.py.
# Dobieramy pakiety wąsko, zamiast brać texlive-full (kilka GB):
# texlive-xetex — silnik XeLaTeX (unicode + fonty systemowe),
# texlive-latex-recommended — geometry, graphicx,
# texlive-fontspec... (w xetex) — wybór fontu systemowego,
# fonts-dejavu-core — polskie znaki ORAZ glify astrologiczne (♄ ♓ ☉),
# librsvg2-bin — rsvg-convert: SVG kosmogramu → PDF.
FROM python:3.12-slim
ENV DEBIAN_FRONTEND=noninteractive
RUN apt-get update && apt-get install -y --no-install-recommends \
texlive-xetex \
texlive-latex-recommended \
fonts-dejavu-core \
librsvg2-bin \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
# Sprawdzenie przy budowie: brak narzędzia ma wyjść TERAZ, nie przy pierwszym
# raporcie na produkcji.
RUN xelatex --version > /dev/null && rsvg-convert --version > /dev/null \
&& echo "xelatex + rsvg-convert OK"
EXPOSE 8004
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8004"]
View File
+77
View File
@@ -0,0 +1,77 @@
"""Kompilacja raportu do PDF (PRE-24).
Dwa kroki zewnętrzne, oba w katalogu tymczasowym, który po sobie sprzątamy:
1. SVG kosmogramu PDF (`rsvg-convert`). LaTeX nie wstawia SVG wprost;
zamiana na PDF zachowuje wektor, więc rysunek zostaje ostry przy dowolnym
powiększeniu.
2. .tex PDF (`xelatex`).
Świadomie NIE logujemy treści raportu to dane urodzeniowe i opisy z baz.
Do logu idzie wyłącznie ogon komunikatów LaTeXa przy błędzie, bo bez niego
diagnoza jest niemożliwa.
"""
from __future__ import annotations
import logging
import pathlib
import shutil
import subprocess
import tempfile
from app.latex import build
log = logging.getLogger("astrololo.render")
WHEEL_SVG = "wheel.svg"
WHEEL_PDF = "wheel.pdf"
DOC = "report"
TIMEOUT = 120 # xelatex na dużym tekście potrzebuje chwili
class RenderError(Exception):
"""Kompilacja się nie udała — z komunikatem nadającym się do pokazania."""
def _run(cmd: list[str], cwd: pathlib.Path) -> subprocess.CompletedProcess:
return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=TIMEOUT)
def _svg_to_pdf(svg: str, work: pathlib.Path) -> str | None:
"""Zwraca nazwę pliku PDF z kosmogramem albo None, gdy nie ma czego rysować."""
if not svg or not svg.strip():
return None
(work / WHEEL_SVG).write_text(svg, encoding="utf-8")
proc = _run(["rsvg-convert", "-f", "pdf", "-o", WHEEL_PDF, WHEEL_SVG], work)
if proc.returncode != 0 or not (work / WHEEL_PDF).exists():
# Brak rysunku nie może wywalić całego raportu — tekst jest ważniejszy.
log.warning("konwersja kosmogramu nie udała się: %s", (proc.stderr or "")[-300:])
return None
return WHEEL_PDF
def to_pdf(report: dict) -> bytes:
"""Raport (dict) → bajty PDF. Rzuca RenderError, gdy LaTeX nie zbuduje pliku."""
with tempfile.TemporaryDirectory(prefix="astrololo-render-") as tmp:
work = pathlib.Path(tmp)
wheel = _svg_to_pdf(report.get("wheel_svg") or "", work)
(work / f"{DOC}.tex").write_text(build(report, wheel), encoding="utf-8")
proc = _run(
["xelatex", "-interaction=nonstopmode", "-halt-on-error", f"{DOC}.tex"],
work,
)
pdf = work / f"{DOC}.pdf"
if not pdf.exists():
tail = (proc.stdout or "")[-1200:]
log.error("xelatex nie zbudował PDF-a; ogon logu: %s", tail)
raise RenderError("LaTeX nie zbudował dokumentu. Ogon logu: " + tail[-400:])
return pdf.read_bytes()
def toolchain_ready() -> dict:
"""Czy narzędzia są na miejscu — do /health, żeby awarię obrazu widzieć od razu,
a nie przy pierwszym raporcie."""
return {
"xelatex": bool(shutil.which("xelatex")),
"rsvg-convert": bool(shutil.which("rsvg-convert")),
}
+131
View File
@@ -0,0 +1,131 @@
"""Budowa dokumentu LaTeX raportu (PRE-24).
Układ wg prośby partnerów:
1. imię i nazwisko,
2. wprowadzone dane,
3. rysunek kosmogramu,
4. interpretacja natalna,
5. predykcje okresowe.
Silnik: XeLaTeX. Nie pdflatex tekst zawiera polskie znaki ORAZ glify
astrologiczne ( ), a te wymagają silnika unicode'owego z fontem systemowym.
DejaVu Sans ma jedno i drugie, dlatego jest fontem głównym.
"""
from __future__ import annotations
# Znaki, które w LaTeXu mają własne znaczenie.
#
# Odwrotnego ukośnika NIE MA na tej liście celowo. Jego zamiennik
# (`\textbackslash{}`) sam zawiera nawiasy klamrowe, więc gdyby szedł w tej samej
# pętli, kolejne podmiany uciekłyby nawiasy, które dopiero co wstawiliśmy —
# i zamiast ukośnika wyszedłby dosłowny tekst `\textbackslash\{\}`. Dlatego
# ukośnik chowamy najpierw pod znacznik zastępczy, a rozwijamy na samym końcu.
_ESCAPES = [
("{", r"\{"),
("}", r"\}"),
("$", r"\$"),
("&", r"\&"),
("#", r"\#"),
("_", r"\_"),
("%", r"\%"),
("~", r"\textasciitilde{}"),
("^", r"\textasciicircum{}"),
]
# Znak spoza tekstu użytkownika — nie da się go wpisać, więc nie da się nim
# oszukać ucieczki.
_BACKSLASH_MARK = "\x00astrololo-backslash\x00"
def esc(text: object) -> str:
"""Ucieka znaki specjalne LaTeXa.
Najgroźniejszy jest `%`: bez ucieczki komentuje RESZTĘ LINII, więc tekst od
modelu urwałby się w środku zdania i nikt by tego nie zauważył PDF
powstałby normalnie, tylko krótszy. Dlatego to nie kosmetyka, a warunek
poprawności raportu.
"""
out = "" if text is None else str(text)
out = out.replace("\\", _BACKSLASH_MARK) # schowaj, żeby nie ucierpiał niżej
for ch, rep in _ESCAPES:
out = out.replace(ch, rep)
return out.replace(_BACKSLASH_MARK, r"\textbackslash{}")
def _paragraphs(text: str) -> str:
"""Tekst od modelu → akapity LaTeXa.
Model oddziela akapity pustą linią. Zamieniamy je na osobne akapity, a same
złamania wiersza w środku akapitu zostawiamy LaTeXowi do złożenia.
"""
blocks = [b.strip() for b in str(text or "").split("\n\n")]
return "\n\n".join(esc(b).replace("\n", " ") for b in blocks if b)
PREAMBLE = r"""\documentclass[11pt,a4paper]{article}
\usepackage{fontspec}
\usepackage[margin=2.4cm]{geometry}
\usepackage{graphicx}
% DejaVu Sans ma i polskie znaki, i glify astrologiczne jeden font na oba
% zadania oszczędza nam dobierania fontu zapasowego.
\setmainfont{DejaVu Sans}
\setlength{\parindent}{0pt}
\setlength{\parskip}{0.6em}
\pagestyle{plain}
\begin{document}
"""
def build(report: dict, wheel_pdf: str | None = None) -> str:
"""Składa źródło .tex. `wheel_pdf` to nazwa pliku z kosmogramem (już PDF)."""
person = str(report.get("person") or "").strip()
data = report.get("data") or {}
natal = report.get("natal") or {}
predictions = report.get("predictions") or []
parts = [PREAMBLE]
# ── 1. imię i nazwisko ──────────────────────────────────────────────
parts.append(r"{\Huge\bfseries " + esc(person or "Raport astrologiczny") + r"\par}")
parts.append(r"\vspace{0.3em}")
# ── 2. wprowadzone dane ─────────────────────────────────────────────
rows = [
("Data urodzenia", data.get("date")),
("Godzina (lokalna)", data.get("time")),
("Offset względem GMT", data.get("tz_offset")),
("Szerokość", data.get("lat")),
("Długość", data.get("lon")),
("Miejsce", data.get("place")),
("System domów", data.get("house_system")),
("Zodiak", data.get("zodiac")),
]
parts.append(r"\begin{tabular}{@{}ll@{}}")
for label, value in rows:
if value not in (None, ""):
parts.append(esc(label) + r" & " + esc(value) + r"\\")
parts.append(r"\end{tabular}")
parts.append(r"\vspace{1em}")
# ── 3. kosmogram ────────────────────────────────────────────────────
if wheel_pdf:
parts.append(r"\begin{center}")
parts.append(r"\includegraphics[width=0.82\textwidth]{" + wheel_pdf + r"}")
parts.append(r"\end{center}")
parts.append(r"\vspace{0.5em}")
# ── 4. interpretacja natalna ────────────────────────────────────────
if natal.get("text"):
parts.append(r"\section*{Interpretacja natalna}")
parts.append(_paragraphs(natal["text"]))
# ── 5. predykcje okresowe ───────────────────────────────────────────
if predictions:
parts.append(r"\section*{Predykcje okresowe}")
for p in predictions:
span = f"{p.get('from_date', '')} {p.get('to_date', '')}"
parts.append(r"\subsection*{" + esc(span) + r"}")
parts.append(_paragraphs(p.get("text")))
parts.append(r"\end{document}")
return "\n".join(parts) + "\n"
+525
View File
@@ -0,0 +1,525 @@
"""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 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"
# Trzecia para: prezentacja ↔ render (PRE-24). Osobny klucz, jak przy pozostałych —
# usługa render dostaje CAŁY raport (dane urodzeniowe + opisy z baz), więc przejęcie
# jej klucza nie może otwierać łącza do logiki ani do danych.
ENV_PRESENTATION_RENDER = "LINK_KEY_PRESENTATION_RENDER"
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 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
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")
def stream_lines(client, url: str, *, payload, headers: dict[str, str] | None = None,
link: Link | None) -> Iterator[str]:
"""Strumieniowe POST zwracające kolejne NIEPUSTE linie NDJSON — na żywo.
Dla okna postępu: linie muszą docierać w trakcie pracy, nie na końcu, więc
czytamy strumień, a nie całe ciało. Gdy łącze ma klucz, żądanie jest
pieczętowane, a odpowiedź odszyfrowywana ramka po ramce; granice ramek NIE
pokrywają się z granicami linii, więc sklejamy bajty w buforze i tniemy je
dopiero na znakach nowej linii.
Bez klucza zachowuje się jak dotąd (surowy strumień), żeby dev bez sekretów
działał bez zmian.
"""
import json as _json
import httpx as _httpx
request_headers = dict(headers or {})
if link is None:
if encryption_required():
# Ten sam kontrakt co w `call`: nie wypuszczamy jawnego żądania, gdy
# szyfrowanie jest wymagane. Bez tego serwer owszem odrzuca (400), ale
# ciało żądania — tu dane urodzenia — zdążyłoby już pójść w eter.
raise LinkError(
f"{ENV_REQUIRED} jest włączone, ale brak klucza łącza — strumień "
f"NIE został wysłany, żeby jego treść nie poszła jawnym tekstem"
)
with client.stream("POST", url, json=payload, headers=request_headers) as response:
response.raise_for_status()
for text_line in response.iter_lines():
if text_line:
yield text_line
return
path = _httpx.URL(url).path
stamp = stamp_now()
body = frame_out(link.seal(REQUEST, path, stamp, 0, _json.dumps(payload).encode("utf-8")))
request_headers.update({HEADER_ENC: VERSION, HEADER_TS: stamp, "Content-Type": CONTENT_TYPE})
with client.stream("POST", url, content=body, headers=request_headers) as response:
response.raise_for_status()
buffer = bytearray()
for plain in open_response_stream(response, link):
buffer += plain
while True:
nl = buffer.find(b"\n")
if nl < 0:
break
text_line = bytes(buffer[:nl])
del buffer[:nl + 1]
if text_line:
yield text_line.decode("utf-8")
if buffer:
yield bytes(buffer).decode("utf-8")
+62
View File
@@ -0,0 +1,62 @@
"""Warstwa RENDER — usługa składająca raport PDF (PRE-24).
OSOBNA usługa, nie część prezentacji, z tego samego powodu co silnik swisseph
(LOG-27): TeX Live waży setki megabajtów. Trzymanie go w obrazie produktu
spowalniałoby każdy build i deploy, a tak aktualizuje się niezależnie i awaria
renderu nie kładzie aplikacji.
Rozmawia TYLKO z prezentacją, po szyfrowanym łączu (własny, trzeci klucz
dostaje cały raport: dane urodzeniowe i opisy z baz). W klastrze wystawiona jako
ClusterIP, bez wejścia z zewnątrz.
Świadomie nie zapisuje niczego na dysk poza katalogiem tymczasowym, który sprząta
po sobie raport nie może zostawiać śladu w kontenerze.
"""
from __future__ import annotations
import logging
from fastapi import FastAPI, HTTPException
from fastapi.responses import Response
from app import link_crypto, security
from app.compile import RenderError, to_pdf, toolchain_ready
log = logging.getLogger("astrololo.render")
app = FastAPI(title="astrololo · usługa render (PDF)")
security.install(app, "render") # 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_RENDER, "render")
@app.post("/pdf")
def build_pdf(report: dict) -> Response:
"""Raport (JSON) → PDF. Wejście: person, data, wheel_svg, natal, predictions.
Zwracamy surowe bajty PDF-a, nie base64 łącze i tak jest szyfrowane, więc
kodowanie tylko rozdmuchałoby odpowiedź o trzecią część.
"""
if not isinstance(report, dict):
raise HTTPException(422, "Oczekiwano obiektu JSON z raportem.")
try:
pdf = to_pdf(report)
except RenderError as e:
raise HTTPException(500, str(e))
except FileNotFoundError as e: # brak xelatex/rsvg w obrazie
raise HTTPException(503, f"Brak narzędzia w obrazie usługi render: {e}")
return Response(
content=pdf,
media_type="application/pdf",
headers={"Content-Disposition": 'attachment; filename="raport.pdf"'},
)
@app.get("/health")
def health() -> dict:
"""Raportuje też obecność narzędzi — dzięki temu zepsuty obraz widać od razu,
a nie dopiero przy pierwszym raporcie."""
tools = toolchain_ready()
return {"status": "ok" if all(tools.values()) else "degraded",
"layer": "render", "tools": tools}
+50
View File
@@ -0,0 +1,50 @@
"""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)
+4
View File
@@ -0,0 +1,4 @@
fastapi>=0.115
uvicorn[standard]>=0.34
# Szyfrowanie łącza z prezentacją (PRE-16/PRE-24): AES-256-GCM + HKDF
cryptography>=44.0
+130
View File
@@ -0,0 +1,130 @@
"""Generowanie źródła LaTeX raportu (PRE-24).
Najważniejsza część to UCIECZKA znaków specjalnych. Kompilacji tu nie
uruchomimy (brak TeX Live w tym środowisku), ale to i tak nie ona jest tu
najbardziej krucha: błędna ucieczka nie wywala kompilacji, tylko po cichu psuje
treść najgorszy możliwy wynik, bo PDF powstaje i wygląda poprawnie.
"""
import pytest
from app.latex import build, esc
# ─────────────────────────── ucieczka znaków specjalnych ────────────────
def test_percent_is_escaped():
"""NAJGROŹNIEJSZY przypadek: `%` bez ucieczki komentuje resztę linii, więc
zdanie urywa się w środku, a PDF powstaje normalnie tylko krótszy."""
out = esc("wzrost o 50% w tym okresie")
assert r"50\%" in out
assert "w tym okresie" in out
def test_backslash_is_escaped_first():
"""Odwrotny ukośnik musi iść pierwszy, inaczej zepsulibyśmy ucieczki
wstawione później (np. `\\%` zamieniłoby się w coś innego)."""
assert esc("a\\b") == r"a\textbackslash{}b"
def test_all_specials_are_escaped():
for ch in ("$", "&", "#", "_", "%", "{", "}"):
assert ch not in esc(ch).replace("\\" + ch, ""), f"{ch} nie zostało uciec­zone"
def test_tilde_and_caret_use_commands():
assert esc("~") == r"\textasciitilde{}"
assert esc("^") == r"\textasciicircum{}"
def test_polish_and_glyphs_pass_through():
"""XeLaTeX bierze unicode wprost — polskich znaków ani glifów nie ruszamy."""
text = "Zażółć gęślą jaźń ♄ ♓ ☉"
assert esc(text) == text
def test_escape_handles_none_and_numbers():
assert esc(None) == ""
assert esc(42) == "42"
# ─────────────────────────────── układ dokumentu ─────────────────────────
def _report(**over):
base = {
"person": "Jan Kowalski",
"data": {"date": "1984-04-30", "time": "11:20", "tz_offset": 2,
"lat": 50.0647, "lon": 19.945, "place": "Kraków",
"house_system": "whole_sign", "zodiac": "tropical"},
"natal": {"text": "Interpretacja natalna.\n\nDrugi akapit."},
"predictions": [
{"from_date": "2026-01-01", "to_date": "2026-03-31", "text": "Pierwszy kwartał."},
{"from_date": "2026-07-01", "to_date": "2026-09-30", "text": "Trzeci kwartał."},
],
}
base.update(over)
return base
def test_document_is_complete():
tex = build(_report())
assert tex.startswith(r"\documentclass")
assert tex.rstrip().endswith(r"\end{document}")
def test_uses_unicode_engine_setup():
"""XeLaTeX + fontspec — bez tego polskie znaki i glify nie wyjdą."""
tex = build(_report())
assert r"\usepackage{fontspec}" in tex
assert r"\setmainfont" in tex
def test_section_order_matches_the_brief():
"""Prośba partnerów: imię → dane → rysunek → natalna → predykcje."""
tex = build(_report(), wheel_pdf="wheel.pdf")
person = tex.index("Jan Kowalski")
data = tex.index("Data urodzenia")
wheel = tex.index("includegraphics")
natal = tex.index("Interpretacja natalna")
preds = tex.index("Predykcje okresowe")
assert person < data < wheel < natal < preds
def test_person_name_opens_the_document():
tex = build(_report())
body = tex.split(r"\begin{document}")[1]
assert body.index("Jan Kowalski") < 80, "imię ma być na samym początku"
def test_all_predictions_are_included():
tex = build(_report())
assert "2026-01-01" in tex and "2026-07-01" in tex
assert "Pierwszy kwartał" in tex and "Trzeci kwartał" in tex
def test_paragraphs_are_preserved():
"""Model oddziela akapity pustą linią — mają zostać akapitami, nie zlepkiem."""
tex = build(_report())
assert "Interpretacja natalna." in tex and "Drugi akapit." in tex
assert "Interpretacja natalna.\n\nDrugi akapit." in tex
def test_hostile_text_cannot_break_the_document():
"""Tekst od modelu jest wejściem z ZEWNĄTRZ — nie może wstrzyknąć polecenia
ani urwać dokumentu."""
nasty = r"100% \end{document} \input{/etc/passwd} $x_1$ #& {}"
tex = build(_report(natal={"text": nasty}))
body = tex.split(r"\begin{document}")[1]
assert body.count(r"\end{document}") == 1, "tekst urwał dokument"
assert r"\input{" not in body
def test_missing_parts_are_simply_absent():
"""Niekompletny materiał ma dać krótszy raport, nie wyjątek."""
tex = build({"person": "", "data": {}})
assert r"\begin{document}" in tex and r"\end{document}" in tex
assert "Raport astrologiczny" in tex # zapas, gdy brak imienia
def test_wheel_is_optional():
assert "includegraphics" not in build(_report())
assert "includegraphics" in build(_report(), wheel_pdf="wheel.pdf")