Files
flowdeck/app/services/draco_urls.py
T
Spark 283e226adf
FlowDeck CI / test (push) Successful in 29m39s
FlowDeck CI / lint (push) Successful in 2m18s
FlowDeck CI / docker (push) Successful in 1m41s
fix(draco): URLs sœurs dérivées de l'environnement de l'agent (7.82.3)
Résolution uniforme des quatre sœurs (ObsiGate, Flowdeck, TikTak,
Kanbante) dans app/services/draco_urls.py : explicite prioritaire
(contrat inchangé), sinon dérivation depuis l'hôte de l'agent connu
via AGENT_EVENEMENTS_URL (agent<SUFFIXE>.dracodev.net →
<slug><SUFFIXE>.dracodev.net), sinon défaut officiel. Hors motif
(autre domaine, LAN, IP) : aucune dérivation. Gitea inchangé.
Tests +4 dans test_draco_urls_devprod.py.
2026-10-26 00:46:21 +00:00

234 lines
7.9 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.
"""
from __future__ import annotations
import os
import re
from urllib.parse import urlparse
from app.config import settings
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$")
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 _resoudre_soeur(env_name: str, settings_attr: str, app: str) -> str:
"""Résolution d'une URL sœur : explicite > dérivation > 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 la dérivation d'environnement. 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
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."""
return _resoudre("DRACO_AGENT_URL", "draco_agent_url", "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``) >
APP_BASE_URL si elle n'est pas locale > 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}"
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 (hôte = config)."""
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}