FabriceandClaude Opus 5.5
4366db94ed
site.toml [administration] adresse : l'accueil affiche un bouton « Connexion » vers l'interface. Celle-ci peut tourner derrière le serveur web du site (rtg interface --proxy --prefixe /admin) : adresses préfixées, mot de passe obligatoire, visiteur identifié par X-Forwarded-For, cookie Secure limité au préfixe quand le site est en HTTPS. Exemples nginx et Apache dans le README. Le bouton Arrêter ne remplace plus le bloc de la tâche en même temps que la relève. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Mf2kATMtqVABHSYdxJMnim
222 lines
11 KiB
Python
222 lines
11 KiB
Python
"""Interface : jours imposés aux traces sans heures, suggestion d'après les
|
|
photos, et quelques échanges avec le serveur local."""
|
|
import os
|
|
import threading
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import date
|
|
from http.server import ThreadingHTTPServer
|
|
|
|
import pytest
|
|
|
|
from rtg import config, sources
|
|
from rtg import acces, corrections
|
|
from rtg.interface import serveur
|
|
from rtg.interface.securite import Gardien
|
|
from test_chaine import _photo, pytestmark # noqa: F401 (mêmes prérequis)
|
|
|
|
|
|
def _etape(chemin, lat0, lon0):
|
|
pts = "".join(f'<trkpt lat="{lat0 + i * 0.01:.5f}" lon="{lon0 + i * 0.012:.5f}"></trkpt>' for i in range(50))
|
|
with open(chemin, "w") as f:
|
|
f.write(f'<?xml version="1.0"?><gpx version="1.1" creator="68degres.fr" '
|
|
f'xmlns="http://www.topografix.com/GPX/1/1"><trk><trkseg>{pts}</trkseg></trk></gpx>')
|
|
|
|
|
|
@pytest.fixture
|
|
def voyage_sans_heures(tmp_path):
|
|
racine = tmp_path / "site"
|
|
voyage = racine / "Ecosse-2027"
|
|
gpx, photos = tmp_path / "gpx", tmp_path / "photos"
|
|
for d in (voyage, gpx, photos):
|
|
d.mkdir(parents=True)
|
|
_etape(gpx / "Etape 1.gpx", 55.9, 3.2)
|
|
_etape(gpx / "Etape 2.gpx", 56.7, 2.24)
|
|
_photo(str(photos / "P1.jpg"), "2027:06:11 10:00:00", gps=(56.75, 2.30))
|
|
config.ecrire_toml(str(racine / "site.toml"), {"titre": "Test"})
|
|
config.ecrire_toml(str(voyage / "voyage.toml"), {
|
|
"titre": "Écosse", "sources": {"traces": [str(gpx)], "videos": [], "photos": [str(photos)]}})
|
|
return racine, voyage
|
|
|
|
|
|
def _cfg(racine, voyage):
|
|
return config.reglages_voyage(str(voyage), config.reglages_site(str(racine)))
|
|
|
|
|
|
def test_jour_impose_et_suggestion(voyage_sans_heures):
|
|
racine, voyage = voyage_sans_heures
|
|
a = sources.analyser(str(voyage), _cfg(racine, voyage))
|
|
assert not a.journees and len(a.non_datees) == 2
|
|
medias = sources.positions_medias(str(voyage), a)
|
|
par_nom = {os.path.basename(f): p for f, p in a.non_datees.items()}
|
|
assert sources.suggerer_jour(par_nom["Etape 2.gpx"], medias) == (date(2027, 6, 11), 1)
|
|
assert sources.suggerer_jour(par_nom["Etape 1.gpx"], medias) == (None, 0)
|
|
|
|
config.modifier_toml(str(voyage / "voyage.toml"), {"dates_traces": {"Etape 1.gpx": "2027-06-10"}})
|
|
a = sources.analyser(str(voyage), _cfg(racine, voyage))
|
|
assert sorted(a.journees) == [date(2027, 6, 10)]
|
|
assert [os.path.basename(f) for f in a.non_datees] == ["Etape 2.gpx"]
|
|
# Le titre et les sources ont survécu à la réécriture du fichier
|
|
assert config.lire_toml(str(voyage / "voyage.toml"))["titre"] == "Écosse"
|
|
|
|
|
|
@pytest.fixture
|
|
def client(voyage_sans_heures):
|
|
racine, voyage = voyage_sans_heures
|
|
app = serveur.Application(str(racine))
|
|
srv = ThreadingHTTPServer(("127.0.0.1", 0), None)
|
|
port = srv.server_address[1]
|
|
gardien = Gardien(port)
|
|
srv.RequestHandlerClass = serveur.creer_gestionnaire(app, port, gardien)
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
|
sid = gardien.ouvrir_session()
|
|
|
|
def appel(chemin, donnees=None, htmx=True, session=True):
|
|
req = urllib.request.Request(f"http://127.0.0.1:{port}{chemin}",
|
|
data=urllib.parse.urlencode(donnees).encode() if donnees else None)
|
|
if htmx:
|
|
req.add_header("HX-Request", "true")
|
|
if session:
|
|
req.add_header("Cookie", f"{gardien.cookie}={sid}")
|
|
try:
|
|
with urllib.request.urlopen(req) as r:
|
|
return r.status, r.read().decode()
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, e.read().decode()
|
|
appel.port, appel.gardien = port, gardien
|
|
yield appel, racine, voyage
|
|
srv.shutdown()
|
|
|
|
|
|
def test_serveur(client):
|
|
appel, racine, voyage = client
|
|
code, html = appel("/")
|
|
assert code == 200 and "Écosse" in html
|
|
code, html = appel("/voyage/analyse?v=Ecosse-2027")
|
|
assert "2 traces à rattacher" in html and "11 juin 2027" in html
|
|
# Répartir : un jour par trace dans l'ordre des noms
|
|
code, html = appel("/voyage/dates-suite", {"v": "Ecosse-2027", "debut": "2027-06-10"})
|
|
assert "2 journées" in html
|
|
assert config.lire_toml(str(voyage / "voyage.toml"))["dates_traces"] == {
|
|
"Etape 1.gpx": "2027-06-10", "Etape 2.gpx": "2027-06-11"}
|
|
# Requête de modification sans l'en-tête htmx : refusée
|
|
code, _ = appel("/voyage/date", {"v": "Ecosse-2027", "fichier": "Etape 1.gpx", "jour": ""}, htmx=False)
|
|
assert code == 403
|
|
# Voyage hors du site : refusé
|
|
code, html = appel("/voyage/analyse?v=../../etc")
|
|
assert "introuvable" in html.lower()
|
|
|
|
|
|
class _SansRedirection(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, *a, **k):
|
|
return None
|
|
|
|
|
|
def test_acces(client, tmp_path, monkeypatch):
|
|
appel, racine, voyage = client
|
|
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config"))
|
|
monkeypatch.setenv("APPDATA", str(tmp_path / "config"))
|
|
base = f"http://127.0.0.1:{appel.port}"
|
|
ouvreur = urllib.request.build_opener(_SansRedirection)
|
|
|
|
def brut(chemin):
|
|
try:
|
|
return ouvreur.open(base + chemin)
|
|
except urllib.error.HTTPError as e:
|
|
return e
|
|
# Sans session : renvoyé vers la connexion, y compris pour les fichiers du site
|
|
for chemin in ("/", "/site/site.toml", "/voyage?v=Ecosse-2027"):
|
|
r = brut(chemin)
|
|
assert r.status == 303 and r.headers["Location"] == "/connexion", chemin
|
|
assert appel("/", session=False)[0] == 200 # htmx : HX-Redirect, corps vide
|
|
# La clé d'accès ouvre une session (cookie HttpOnly), une fausse clé non
|
|
r = brut("/?jeton=" + appel.gardien.jeton)
|
|
assert r.status == 303 and "HttpOnly" in r.headers["Set-Cookie"] and "SameSite=Strict" in r.headers["Set-Cookie"]
|
|
assert brut("/?jeton=faux").headers["Location"] == "/connexion"
|
|
# Mot de passe : faux, juste, puis blocage après 5 échecs
|
|
acces.definir("un mot de passe solide")
|
|
code, html = appel("/connexion", {"mot_de_passe": "non"}, session=False)
|
|
assert "incorrect" in html
|
|
ok, _ = appel.gardien.essayer("10.0.0.1", "un mot de passe solide")
|
|
assert ok
|
|
for _ in range(5):
|
|
appel.gardien.essayer("10.0.0.2", "faux")
|
|
ok, message = appel.gardien.essayer("10.0.0.2", "un mot de passe solide")
|
|
assert not ok and "Trop d'essais" in message
|
|
# En-têtes de sécurité
|
|
r = urllib.request.urlopen(urllib.request.Request(base + "/connexion"))
|
|
assert r.headers["X-Frame-Options"] == "DENY" and "frame-ancestors 'none'" in r.headers["Content-Security-Policy"]
|
|
|
|
|
|
def test_corrections(client):
|
|
appel, racine, voyage = client
|
|
# Médias : une photo localisée, une sans position
|
|
with open(voyage / "export_photos.csv", "w") as f:
|
|
f.write('"P1.jpg",2027:06:11 10:00:00,56.75,2.30,\n"P2.jpg",2027:06:11 11:00:00,,,\n')
|
|
code, texte = appel("/voyage/position", {"v": "Ecosse-2027", "fichier": "P2.jpg", "lat": "56.8", "lon": "2.36"})
|
|
assert '"deplace": true' in texte
|
|
cfg = config.reglages_voyage(str(voyage), config.reglages_site(str(racine)))
|
|
assert corrections.positions(cfg) == {"P2.jpg": (56.8, 2.36)}
|
|
# Zone masquée du voyage, puis zone privée du site
|
|
appel("/voyage/masque", {"v": "Ecosse-2027", "action": "ajouter", "lat": "56.7", "lon": "2.24", "rayon": "3000"})
|
|
appel("/voyage/masque", {"v": "Ecosse-2027", "action": "ajouter", "portee": "site",
|
|
"lat": "48.8", "lon": "2.1", "rayon": "500"})
|
|
cfg = config.reglages_voyage(str(voyage), config.reglages_site(str(racine)))
|
|
assert corrections.masques(cfg) == [(56.7, 2.24, 3000.0), (48.8, 2.1, 500.0)]
|
|
# Génération : la photo placée apparaît, le début d'Etape 2 est retiré
|
|
config.modifier_toml(str(voyage / "voyage.toml"), {"dates_traces": {"Etape 1.gpx": "2027-06-10",
|
|
"Etape 2.gpx": "2027-06-11"}})
|
|
from rtg import pages, sources
|
|
cfg = config.reglages_voyage(str(voyage), config.reglages_site(str(racine)))
|
|
sources.ranger(str(voyage), cfg, sources.analyser(str(voyage), cfg))
|
|
v = pages.Voyage(str(voyage), cfg)
|
|
assert set(v.df_p["Fichier"]) == {"P1.jpg", "P2.jpg"}
|
|
v.lire_traces()
|
|
assert v.points_masques > 0
|
|
for pistes in v.traces.values():
|
|
for piste, pts in pistes:
|
|
assert all(corrections.distance_m(la, lo, 56.7, 2.24) > 3000 for la, lo in pts)
|
|
|
|
|
|
def test_derriere_le_serveur_web(voyage_sans_heures, tmp_path, monkeypatch):
|
|
"""--proxy --prefixe /admin : adresses préfixées, mot de passe, cookie Secure."""
|
|
racine, voyage = voyage_sans_heures
|
|
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config"))
|
|
monkeypatch.setenv("APPDATA", str(tmp_path / "config"))
|
|
acces.definir("un mot de passe solide")
|
|
app = serveur.Application(str(racine), "/admin")
|
|
srv = ThreadingHTTPServer(("127.0.0.1", 0), None)
|
|
port = srv.server_address[1]
|
|
gardien = Gardien(port, proxy=True, prefixe=app.prefixe)
|
|
srv.RequestHandlerClass = serveur.creer_gestionnaire(app, port, gardien)
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
|
ouvreur = urllib.request.build_opener(_SansRedirection)
|
|
entetes = {"Host": "roadtrip.example", "X-Forwarded-For": "203.0.113.7", "X-Forwarded-Proto": "https"}
|
|
|
|
def requete(chemin, donnees=None, extra=None):
|
|
req = urllib.request.Request(f"http://127.0.0.1:{port}{chemin}", headers=dict(entetes, **(extra or {})),
|
|
data=urllib.parse.urlencode(donnees).encode() if donnees else None)
|
|
try:
|
|
return ouvreur.open(req)
|
|
except urllib.error.HTTPError as e:
|
|
return e
|
|
try:
|
|
r = requete("/admin/")
|
|
assert r.status == 303 and r.headers["Location"] == "/admin/connexion"
|
|
assert requete("/").status == 404 # hors du préfixe
|
|
assert requete("/admin/connexion").status == 200
|
|
r = requete("/admin/connexion", {"mot_de_passe": "un mot de passe solide"}, {"HX-Request": "true"})
|
|
cookie = r.headers["Set-Cookie"]
|
|
assert r.headers["HX-Redirect"] == "/admin/" and "Path=/admin/" in cookie and "Secure" in cookie
|
|
sid = cookie.split(";")[0]
|
|
r = requete("/admin/", extra={"Cookie": sid})
|
|
html = r.read().decode()
|
|
assert r.status == 200 and 'href="/admin/nouveau"' in html and 'href="/"' in html # Voir le site
|
|
# Les échecs sont comptés par visiteur (X-Forwarded-For), pas pour le serveur web
|
|
for _ in range(5):
|
|
requete("/admin/connexion", {"mot_de_passe": "faux"}, {"HX-Request": "true"})
|
|
assert gardien.attente("203.0.113.7") > 0 and gardien.attente("127.0.0.1") == 0
|
|
finally:
|
|
srv.shutdown()
|
|
serveur.ENV.globals["base"] = ""
|