Files
flowdeck/app/services/draco_urls.py
T
Spark fd79a49d63
FlowDeck CI / test (push) Failing after 3h13m27s
FlowDeck CI / lint (push) Successful in 1m47s
FlowDeck CI / docker (push) Skipped
fix(draco): résolution dev/prod par l'hôte de la requête (7.82.4)
Le mode console du panneau restait une page morte sur flowdeck-dev :
CSP frame-src/connect-src et DRACO_PANNEAU_CONFIG.agent_url pointaient
l'agent prod (pas encore installée). Deux trous de la 7.82.3 :
agent_url() sans aucune dérivation, et aucune résolution lisant
l'hôte de la requête.

Pour tout ce qui est rendu à une requête (barre/DRACO_URLS, iframe
du panneau, CSP, boutons et redirections SSO) : explicite >
hôte public de la requête (motif fermé
^(og|flowdeck|tiktak|kanbante|agent)(SUFFIXE).dracodev.net$,
suffixe appliqué aux sœurs et à l'agent) > dérivation par les
adresses de l'agent connues en config (élargie : agent_url() se
dérive de l'URL d'événements explicite) > défaut officiel.
L'hôte est posé par le middleware CSP dans un ContextVar (patron
du nonce CSP) : iframe et CSP de la même réponse consomment la
même origine résolue. Contextes sans requête inchangés sinon.

Tests : +12 dans test_draco_urls_devprod.py (25/25) — Host dev
sans aucun env → tout en -dev ; Host prod → officiels ; Host LAN
→ repli config/défauts ; explicite gagne sur l'hôte ; motif fermé.
Suite complète : 4 failed / 26 errors identiques sur arbre propre
(familles préexistantes MCP/realtime/my_tasks/gallery/sync).
2026-10-26 03:04:50 +00:00

376 lines
14 KiB
Python

"""URLs publiques des applications dracodev — résolution dev/prod.
Source unique pour tout le code vivant qui doit nommer une autre
application (CSP, panneau Spark, barre commune, SSO, presets MCP,
références croisées…). Règle du chantier dev/prod : **aucun nom
public codé en dur ailleurs** ; les défauts ci-dessous sont les noms
OFFICIELS (production) et un environnement dev déclare ses noms en
``-dev`` dans le ``.env`` (qui survit aux déploiements par release,
car il est passé par ``environment:``/``env_file:``) :
DRACO_SSO_URL=https://og-dev.dracodev.net
DRACO_AGENT_URL=https://agent-dev.dracodev.net
TIKTAK_URL=https://tiktak-dev.dracodev.net
KANBANTE_URL=https://kanbante-dev.dracodev.net
FLOWDECK_URL=https://flowdeck-dev.dracodev.net (ou OAUTH_REDIRECT_URI)
Priorité de résolution, à chaque appel (les tests et les modules qui
lisaient déjà ``os.environ`` au moment de l'appel gardent ce
comportement) : variable d'environnement > ``app.config.settings`` >
défaut officiel. Aucun défaut ne contient ``-dev``.
**Dérivation d'environnement (v7.82.3)** — pour les quatre URL
sœurs (ObsiGate, Flowdeck, TikTak, Kanbante), entre la configuration
explicite et le défaut officiel : si l'origine de l'agent telle que
l'app la connaît déjà (``AGENT_EVENEMENTS_URL``, via
``agent_evenements_url()``) a un hôte de la forme
``agent<SUFFIXE>.dracodev.net`` (SUFFIXE éventuellement vide,
typiquement ``-dev``), l'URL sœur est dérivée :
``https://<slug><SUFFIXE>.dracodev.net`` (slugs : ``og``,
``flowdeck``, ``tiktak``, ``kanbante``). Ainsi une installation dev
qui ne déclare que l'agent pointe ses liens inter-apps en ``-dev``
sans quatre variables de plus. Si l'hôte de l'agent ne correspond
pas au motif (autre domaine, LAN, IP), **aucune dérivation** : on
ne fabrique jamais un lien deviné. L'URL Gitea ne change pas.
**Résolution par l'hôte de la requête (v7.82.4)** — la v7.82.3
laissait deux trous, prouvés en direct sur flowdeck-dev (CSP
``frame-src 'self' https://agent.dracodev.net`` alors que la prod
n'existe pas encore → console du panneau = page morte) :
1. ``agent_url()`` n'avait **aucune** dérivation : même avec
``AGENT_EVENEMENTS_URL`` en ``-dev``, l'agent lui-même retombait
sur le nom officiel ;
2. **aucune** résolution ne lisait l'hôte de la requête : un
``.env`` dev sans ``DRACO_AGENT_URL`` ni ``AGENT_EVENEMENTS_URL``
retombait entièrement sur la prod.
Pour tout ce qui est **rendu à une requête** (liens de la barre /
``DRACO_URLS`` servi, source de l'iframe du panneau, origine agent
des CSP ``frame-src``/``connect-src``, boutons et redirections
SSO), l'ordre de résolution devient :
1. variable explicite de l'installation — prioritaire, inchangée ;
2. **hôte public de la requête** (posé par le middleware CSP dans
``HOTE_REQUETE``, même patron que le nonce CSP) : motif fermé
``^(og|flowdeck|tiktak|kanbante|agent)(<SUFFIXE>)\\.dracodev\\.net$``
(suffixe optionnel, p. ex. ``-dev``, vide en prod) →
l'environnement = ce suffixe, appliqué à toutes les sœurs **et
à l'agent**. Jamais de suffixe inventé depuis un hôte arbitraire ;
hors motif (LAN, IP) → étape suivante ;
3. dérivation par les adresses de l'agent connues en configuration
(URL d'événements **ou** base du service) — aussi la résolution
des contextes **sans requête**, inchangée sinon ;
4. défaut officiel.
L'iframe du panneau et la CSP de la même réponse consomment la
même origine résolue : les deux lisent ``agent_url()`` pendant la
même requête, avec le même ``HOTE_REQUETE``.
"""
from __future__ import annotations
import os
import re
from contextvars import ContextVar
from urllib.parse import urlparse
from app.config import settings
# Hôte public de la requête en cours (``X-Forwarded-Host`` sinon
# ``Host``), posé par le middleware CSP avant le traitement — même
# patron que ``CSP_NONCE`` : le contexte est copié dans la tâche qui
# rend les templates. Vide hors requête (tâches de fond, CLI, tests
# unitaires) → la résolution retombe sur la configuration.
HOTE_REQUETE: ContextVar[str] = ContextVar("draco_hote_requete", default="")
def definir_hote_requete(hote: str) -> None:
"""Mémorise l'hôte public de la requête (premier de la liste)."""
HOTE_REQUETE.set((hote or "").split(",")[0].strip().lower())
OFFICIELLES = {
"obsigate": "https://og.dracodev.net",
"agent": "https://agent.dracodev.net",
"tiktak": "https://tiktak.dracodev.net",
"kanbante": "https://kanbante.dracodev.net",
"flowdeck": "https://flowdeck.dracodev.net",
"gitea": "https://git.dracodev.net",
}
def _nettoyer(url: str) -> str:
return (url or "").strip().rstrip("/")
def _resoudre(env_name: str, settings_attr: str, app: str) -> str:
env = _nettoyer(os.environ.get(env_name, ""))
if env:
return env
cfg = _nettoyer(getattr(settings, settings_attr, "") or "")
if cfg:
return cfg
return OFFICIELLES[app]
# Slugs d'hôte des applications sœurs (dérivation d'environnement).
_SLUGS_SOEURS = {
"obsigate": "og",
"flowdeck": "flowdeck",
"tiktak": "tiktak",
"kanbante": "kanbante",
}
# Hôte de l'agent : agent<SUFFIXE>.dracodev.net (SUFFIXE = ``-dev``…).
_MOTIF_HOTE_AGENT = re.compile(r"^agent(-[a-z0-9-]+)?\.dracodev\.net$")
# Hôte public d'une requête : motif FERMÉ des cinq applications.
# Le suffixe capturé (groupe 2, p. ex. ``-dev``) est l'environnement
# de TOUTE l'installation — jamais un suffixe inventé depuis un
# hôte arbitraire : hors motif (autre domaine, LAN, IP), None.
_MOTIF_HOTE_PUBLIC = re.compile(
r"^(og|flowdeck|tiktak|kanbante|agent)(-[a-z0-9-]+)?\.dracodev\.net$")
# Slugs pour la résolution par l'hôte de la requête (l'agent inclus).
_SLUGS_TOUTES = {**_SLUGS_SOEURS, "agent": "agent"}
def _suffixe_environnement() -> str | None:
"""Suffixe d'environnement porté par l'hôte de l'agent, ou None.
Source : l'origine de l'agent telle que l'app la connaît déjà —
``agent_evenements_url()`` (``AGENT_EVENEMENTS_URL`` explicite,
sinon dérivée de la base de l'agent). Retourne ``""`` pour
l'hôte officiel, ``"-dev"`` pour ``agent-dev.dracodev.net``, et
None si l'hôte ne correspond pas au motif (aucune dérivation).
"""
hote_agent = (urlparse(agent_evenements_url()).hostname or "").lower()
m = _MOTIF_HOTE_AGENT.match(hote_agent)
if not m:
return None
return m.group(1) or ""
def _url_derivee(app: str) -> str:
"""URL sœur dérivée de l'environnement de l'agent ("" si aucune)."""
slug = _SLUGS_SOEURS.get(app)
if not slug:
return ""
suffixe = _suffixe_environnement()
if suffixe is None:
return ""
return f"https://{slug}{suffixe}.dracodev.net"
def _suffixe_requete() -> str | None:
"""Suffixe d'environnement porté par l'hôte de la requête, ou None.
Motif fermé ``_MOTIF_HOTE_PUBLIC`` sur l'hôte public mémorisé
(port retiré) : ``""`` pour un nom officiel, ``"-dev"`` pour un
nom en ``-dev``, None hors motif ou hors requête.
"""
hote_req = HOTE_REQUETE.get().split(":")[0]
if not hote_req:
return None
m = _MOTIF_HOTE_PUBLIC.match(hote_req)
if not m:
return None
return m.group(2) or ""
def _url_depuis_requete(app: str) -> str:
"""URL d'``app`` dérivée de l'hôte de la requête ("" si aucune)."""
slug = _SLUGS_TOUTES.get(app)
if not slug:
return ""
suffixe = _suffixe_requete()
if suffixe is None:
return ""
return f"https://{slug}{suffixe}.dracodev.net"
def _suffixe_evenements_explicite() -> str | None:
"""Suffixe porté par l'URL d'événements **explicite**, ou None.
Lit ``AGENT_EVENEMENTS_URL`` (environnement puis settings) sans
passer par ``agent_evenements_url()`` : pas de récursion avec
``agent_url()``, dont c'est une source de dérivation.
"""
for candidate in (os.environ.get("AGENT_EVENEMENTS_URL", ""),
getattr(settings, "agent_evenements_url", "") or ""):
candidate = _nettoyer(candidate)
if candidate:
hote_agent = (urlparse(candidate).hostname or "").lower()
m = _MOTIF_HOTE_AGENT.match(hote_agent)
return (m.group(1) or "") if m else None
return None
def _resoudre_soeur(env_name: str, settings_attr: str, app: str) -> str:
"""Résolution d'une URL sœur, dans l'ordre du chantier :
explicite > **hôte de la requête** > dérivation par l'agent
connu en configuration > officiel.
La valeur ``settings`` égale au nom officiel est le *défaut* de
``app/config.py``, pas un choix explicite : elle ne doit pas
bloquer les dérivations. Une valeur settings différente (dev,
LAN, domaine propre) reste prioritaire.
"""
env = _nettoyer(os.environ.get(env_name, ""))
if env:
return env
cfg = _nettoyer(getattr(settings, settings_attr, "") or "")
if cfg and cfg != OFFICIELLES[app]:
return cfg
depuis_requete = _url_depuis_requete(app)
if depuis_requete:
return depuis_requete
derivee = _url_derivee(app)
if derivee:
return derivee
if cfg:
return cfg
return OFFICIELLES[app]
def obsigate_url() -> str:
"""ObsiGate — aussi l'émettrice SSO (DRACO_SSO_URL, déjà en service)."""
return _resoudre_soeur("DRACO_SSO_URL", "draco_sso_url", "obsigate")
def agent_url() -> str:
"""Service d'agents (Spark) — base sans chemin.
Même ordre que les sœurs (la v7.82.3 l'avait laissé sans aucune
dérivation — trou n° 1 du mode console mort en dev) : explicite
(``DRACO_AGENT_URL`` / settings non officiel) > hôte de la
requête > dérivation par l'URL d'événements explicite
(``AGENT_EVENEMENTS_URL`` : ``agent-dev…`` → base ``agent-dev…``)
> défaut officiel.
"""
env = _nettoyer(os.environ.get("DRACO_AGENT_URL", ""))
if env:
return env
cfg = _nettoyer(getattr(settings, "draco_agent_url", "") or "")
if cfg and cfg != OFFICIELLES["agent"]:
return cfg
depuis_requete = _url_depuis_requete("agent")
if depuis_requete:
return depuis_requete
suffixe = _suffixe_evenements_explicite()
if suffixe is not None:
return f"https://agent{suffixe}.dracodev.net"
if cfg:
return cfg
return OFFICIELLES["agent"]
def tiktak_url() -> str:
return _resoudre_soeur("TIKTAK_URL", "tiktak_url", "tiktak")
def kanbante_url() -> str:
return _resoudre_soeur("KANBANTE_URL", "kanbante_url", "kanbante")
def gitea_url() -> str:
return _resoudre("GITEA_URL", "gitea_url", "gitea")
def flowdeck_url() -> str:
"""URL publique de cette instance de Flowdeck.
FLOWDECK_URL > origine de OAUTH_REDIRECT_URI (déjà déclarée en
dev : ``https://flowdeck-dev.dracodev.net/auth/callback``) >
hôte de la requête (motif fermé) > APP_BASE_URL si elle n'est
pas locale > dérivation par l'agent > nom officiel.
"""
env = _nettoyer(os.environ.get("FLOWDECK_URL", ""))
if env:
return env
cfg = _nettoyer(getattr(settings, "flowdeck_url", "") or "")
if cfg:
return cfg
for candidate in (
os.environ.get("OAUTH_REDIRECT_URI", ""),
getattr(settings, "oauth_redirect_uri", "") or "",
):
candidate = _nettoyer(candidate)
if candidate:
p = urlparse(candidate)
if p.scheme and p.netloc:
return f"{p.scheme}://{p.netloc}"
# Étape 2 du chantier : l'hôte de la requête — c'est l'origine
# que le navigateur utilise déjà pour cette instance, donc
# aussi celle du callback SSO (``/auth/draco``) quand rien
# d'explicite n'est déclaré.
depuis_requete = _url_depuis_requete("flowdeck")
if depuis_requete:
return depuis_requete
base = _nettoyer(getattr(settings, "app_base_url", "") or "")
if base and "localhost" not in base and "127.0.0.1" not in base:
return base
derivee = _url_derivee("flowdeck")
if derivee:
return derivee
return OFFICIELLES["flowdeck"]
def agent_evenements_url() -> str:
"""Endpoint du journal d'événements du service d'agents.
AGENT_EVENEMENTS_URL (déjà en service) prime ; sinon dérivée de
la base de l'agent — jamais un nom figé.
"""
env = _nettoyer(os.environ.get("AGENT_EVENEMENTS_URL", ""))
if env:
return env
cfg = _nettoyer(getattr(settings, "agent_evenements_url", "") or "")
if cfg:
return cfg
return f"{agent_url()}/api/v1/evenements"
def origine(url: str) -> str:
"""``scheme://netloc`` d'une URL (pour la CSP)."""
p = urlparse(url or "")
if p.scheme and p.netloc:
return f"{p.scheme}://{p.netloc}"
return _nettoyer(url)
def hote(url: str) -> str:
return (urlparse(url or "").hostname or "").lower()
def toutes() -> dict[str, str]:
"""Toutes les URL résolues — injectées aux templates.
Résolution à l'appel : dans une requête, l'hôte public de la
requête fait foi après les variables explicites ; hors requête,
la configuration seule (voir le docstring du module).
"""
return {
"obsigate": obsigate_url(),
"agent": agent_url(),
"tiktak": tiktak_url(),
"kanbante": kanbante_url(),
"flowdeck": flowdeck_url(),
"gitea": gitea_url(),
}
def hotes_connus(app: str) -> set[str]:
"""Hôtes reconnus pour ``app`` : l'hôte configuré **et** l'hôte
officiel. Les références croisées déjà stockées portent des URL
absolues (officielles aujourd'hui) : elles doivent rester
reconnaissables dans un environnement dev, et inversement —
aucune migration de données n'est faite pour un changement de nom.
"""
resolues = toutes()
hotes = {hote(OFFICIELLES[app])}
if app in resolues:
hotes.add(hote(resolues[app]))
return {h for h in hotes if h}