duffyduckandClaude Opus 5
2c60a19ed6
oauth_get_token hat den access_token im Klartext als Tool-Ergebnis
zurueckgegeben. Tool-Ergebnisse sind Teil der Unterhaltung — der Token ging
damit an Anthropic und lag je nach Konto-Einstellung 30 Tage oder 5 Jahre dort.
Neu: das Brain fuehrt den API-Aufruf SELBST aus.
- `oauth_call(service, url, method, body)` haengt die Authorization an,
schickt den Request und gibt nur die Antwort zurueck. Der Token verlaesst
den Brain-Prozess nicht.
- `oauth_get_token` sagt nur noch, OB authentifiziert ist und wie lange noch.
Refresh laeuft weiter automatisch.
- Skills bleiben unveraendert: die holen den Token serverseitig ueber
{BRAIN_INTERNAL_URL}/oauth/<service>/token — der laeuft nie durch ein
Gespraech.
Host-Sperre als Sicherheitskern: die URL kommt vom Modell, also koennte ein
Aufruf den Token sonst an einen beliebigen Server schicken. Erlaubt sind nur
Hosts, die zum Provider gehoeren (aus auth_url/token_url, plus optionale
api_hosts). Verglichen wird auf den letzten zwei Labels, NICHT per endswith —
"boesespotify.com" endet sonst auch auf "spotify.com". Dazu: nur https (sonst
ginge der Token im Klartext ueber die Leitung) und nur die ueblichen Methoden.
19 Pruefungen, darunter die Angriffsvarianten: boesespotify.com,
spotify.com.angreifer.de, accounts.spotify.com.evil.io, spotify.co, leerer
Host — alle blockiert; api.spotify.com und abschliessender Punkt erlaubt.
System-Prompt und seed_rules nachgezogen, sonst erklaert ARIA sich das alte
Verhalten selbst. Beide sagen jetzt zusaetzlich ausdruecklich: Secrets gehoeren
nie in den Chat, und ein Secret nie in einer Antwort wiederholen. Genau so ist
hier naemlich ein client_secret abgeflossen — nicht ueber ein Tool, sondern
weil es eingetippt und dann zurueckzitiert wurde.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
552 lines
23 KiB
Python
552 lines
23 KiB
Python
"""
|
|
OAuth-Manager fuer ARIA. Generischer OAuth2 Authorization-Code-Flow fuer
|
|
Spotify, Google, GitHub, Strava, Microsoft etc.
|
|
|
|
Architektur:
|
|
- Brain haelt einen Pending-Store: state-String → pending Auth-Request
|
|
(mit timeout). Wenn ein Callback ankommt (via aria-bridge ueber RVS),
|
|
matched der state und der code wird gegen access_token getauscht.
|
|
- Token-Storage: /shared/config/oauth_tokens.json (pro Service ein Eintrag
|
|
mit access_token, refresh_token, expires_at, scope).
|
|
- Provider-Configs: /shared/config/oauth_apps.json — pro Service
|
|
{client_id, client_secret, auth_url, token_url, scopes, ...}. Wird
|
|
typischerweise via Diagnostic-UI gefuellt.
|
|
- Token-Refresh: automatisch wenn access_token abgelaufen oder < 60s
|
|
bis Ablauf bei get_token() Aufruf.
|
|
|
|
OAuth-Callback-URL: https://{RVS_HOST}:{RVS_PORT_PUBLIC}/oauth/callback/{service}
|
|
RVS_PORT_PUBLIC ist nicht zwingend gleich RVS_PORT (port-mapping via TLS-Proxy).
|
|
ARIA setzt die URL beim Auth-Request automatisch — Stefan muss sie EINMAL pro
|
|
Service im Provider-Dashboard registrieren.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import logging
|
|
import os
|
|
import secrets
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CONFIG_DIR = Path("/shared/config")
|
|
APPS_FILE = CONFIG_DIR / "oauth_apps.json"
|
|
TOKENS_FILE = CONFIG_DIR / "oauth_tokens.json"
|
|
|
|
# Default-Provider-Configs. Werden von oauth_apps.json gemergt (User-Config
|
|
# uebersteuert). Aktuell nur Spotify als out-of-the-box Service — fuer alles
|
|
# andere benutzt ARIA das `oauth_register_provider` Tool (legt Provider on-
|
|
# demand mit den jeweiligen Endpunkten an). Stefan muss bei jedem Provider
|
|
# danach nur client_id + client_secret in Diagnostic / App eintragen.
|
|
DEFAULT_PROVIDERS: dict[str, dict] = {
|
|
"spotify": {
|
|
"auth_url": "https://accounts.spotify.com/authorize",
|
|
"token_url": "https://accounts.spotify.com/api/token",
|
|
"scopes": ["user-read-playback-state", "user-modify-playback-state",
|
|
"user-read-currently-playing", "playlist-read-private",
|
|
"user-library-read"],
|
|
"client_auth": "basic", # client_id:client_secret als Basic-Auth-Header
|
|
},
|
|
}
|
|
|
|
# Pending Auth-Requests: state → {service, scopes, redirect_uri, created_at}
|
|
_PENDING: dict[str, dict] = {}
|
|
PENDING_TTL_SEC = 600 # 10 min — laenger nicht sinnvoll, OAuth-Codes sind eh kurzlebig
|
|
|
|
|
|
# ── Helpers ─────────────────────────────────────────────────
|
|
|
|
|
|
def _callback_url(service: str) -> str:
|
|
"""Baut die Redirect-URL die wir bei der Provider-Auth angeben.
|
|
Liest RVS_HOST / RVS_PORT_PUBLIC / RVS_TLS aus env."""
|
|
host = os.environ.get("RVS_HOST", "").strip()
|
|
if not host:
|
|
raise RuntimeError("RVS_HOST nicht gesetzt — OAuth-Callbacks nicht moeglich")
|
|
port = os.environ.get("RVS_PORT_PUBLIC", os.environ.get("RVS_PORT", "443")).strip()
|
|
tls = os.environ.get("RVS_TLS", "true").strip().lower() != "false"
|
|
scheme = "https" if tls else "http"
|
|
# Default-Ports 443/80 nicht in URL anhaengen
|
|
if (tls and port == "443") or (not tls and port == "80"):
|
|
return f"{scheme}://{host}/oauth/callback/{service}"
|
|
return f"{scheme}://{host}:{port}/oauth/callback/{service}"
|
|
|
|
|
|
def _load_json(path: Path) -> dict:
|
|
try:
|
|
if path.exists():
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception as exc:
|
|
logger.warning("OAuth-Datei %s lesen fehlgeschlagen: %s", path, exc)
|
|
return {}
|
|
|
|
|
|
def _save_json(path: Path, data: dict) -> None:
|
|
try:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
|
tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
tmp.replace(path)
|
|
# 600 — enthaelt Secrets
|
|
try: os.chmod(path, 0o600)
|
|
except OSError: pass
|
|
except Exception as exc:
|
|
logger.error("OAuth-Datei %s speichern fehlgeschlagen: %s", path, exc)
|
|
|
|
|
|
def _provider_config(service: str) -> dict:
|
|
"""Mergt Default-Provider-Config mit User-Override aus oauth_apps.json."""
|
|
defaults = DEFAULT_PROVIDERS.get(service, {}).copy()
|
|
apps = _load_json(APPS_FILE)
|
|
user = (apps.get(service) or {}).copy()
|
|
# Tiefes Merge nicht noetig — die kollidierenden Felder sind alle scalar/list.
|
|
merged = {**defaults, **user}
|
|
return merged
|
|
|
|
|
|
def _provider_credentials(service: str) -> tuple[str, str]:
|
|
"""Liest client_id + client_secret aus oauth_apps.json. Wirft wenn nicht
|
|
konfiguriert — der OAuth-Flow kann ohne nicht starten."""
|
|
apps = _load_json(APPS_FILE)
|
|
entry = apps.get(service) or {}
|
|
cid = (entry.get("client_id") or "").strip()
|
|
sec = (entry.get("client_secret") or "").strip()
|
|
if not cid or not sec:
|
|
raise RuntimeError(
|
|
f"OAuth-App '{service}' nicht konfiguriert. Bitte in Diagnostic > "
|
|
f"OAuth-Apps client_id + client_secret eintragen."
|
|
)
|
|
return cid, sec
|
|
|
|
|
|
def register_provider(service: str, auth_url: str, token_url: str,
|
|
scopes: Optional[list[str]] = None,
|
|
client_auth: str = "body",
|
|
extra_auth_params: Optional[dict] = None,
|
|
accept_header: Optional[str] = None,
|
|
api_hosts: Optional[list[str]] = None) -> dict:
|
|
"""Schreibt einen neuen Provider-Eintrag in oauth_apps.json. KEINE
|
|
Credentials hier — die bleiben Stefans Job (Diagnostic / App-UI). Wird
|
|
vom Brain-Tool `oauth_register_provider` gerufen.
|
|
|
|
Wenn der Service schon existiert: URLs/Scopes werden ueberschrieben,
|
|
aber vorhandene client_id/client_secret bleiben unberuehrt.
|
|
"""
|
|
svc = (service or "").strip()
|
|
if not svc or not all(c.isalnum() or c in "_-" for c in svc) or len(svc) > 60:
|
|
raise ValueError(f"Ungueltiger service-Name: {service!r}")
|
|
if not auth_url.startswith(("http://", "https://")):
|
|
raise ValueError(f"auth_url muss http(s):// sein: {auth_url!r}")
|
|
if not token_url.startswith(("http://", "https://")):
|
|
raise ValueError(f"token_url muss http(s):// sein: {token_url!r}")
|
|
if client_auth not in ("body", "basic"):
|
|
raise ValueError(f"client_auth muss 'body' oder 'basic' sein: {client_auth!r}")
|
|
|
|
apps = _load_json(APPS_FILE)
|
|
entry = apps.get(svc) or {}
|
|
entry["auth_url"] = auth_url.strip()
|
|
entry["token_url"] = token_url.strip()
|
|
if scopes is not None:
|
|
entry["scopes"] = list(scopes)
|
|
entry["client_auth"] = client_auth
|
|
if extra_auth_params is not None:
|
|
entry["extra_auth_params"] = extra_auth_params
|
|
if accept_header is not None:
|
|
entry["accept_header"] = accept_header
|
|
if api_hosts is not None:
|
|
# Zusaetzliche API-Hosts, an die oauth_call den Token schicken darf.
|
|
# Noetig, wenn die API auf einer anderen Domain liegt als der Login
|
|
# (z.B. github.com -> api.github.com waere noch dieselbe Basis, aber
|
|
# manche Provider nutzen voellig andere Namen).
|
|
entry["api_hosts"] = [str(h).strip().lower() for h in api_hosts if str(h).strip()]
|
|
apps[svc] = entry
|
|
_save_json(APPS_FILE, apps)
|
|
logger.info("[oauth] Provider '%s' registriert (auth=%s, token=%s, scopes=%d)",
|
|
svc, auth_url, token_url, len(entry.get("scopes") or []))
|
|
return entry
|
|
|
|
|
|
def _cleanup_pending() -> None:
|
|
"""Entfernt abgelaufene Pending-Auths."""
|
|
now = time.time()
|
|
for state, info in list(_PENDING.items()):
|
|
if now - info.get("created_at", 0) > PENDING_TTL_SEC:
|
|
_PENDING.pop(state, None)
|
|
|
|
|
|
# ── Authorize ───────────────────────────────────────────────
|
|
|
|
|
|
def build_authorize_url(service: str, scopes: Optional[list[str]] = None,
|
|
extra_params: Optional[dict] = None) -> dict:
|
|
"""Baut die Authorize-URL fuer einen Provider. Speichert den state
|
|
im Pending-Store. Returns {url, state, redirect_uri, service}.
|
|
|
|
Wird vom Brain-Tool oauth_authorize gerufen. ARIA gibt die url an Stefan,
|
|
der oeffnet sie im Browser, autorisiert, Provider redirected zur
|
|
redirect_uri (= RVS), RVS broadcasted, bridge forwarded, brain matched
|
|
state → exchange.
|
|
"""
|
|
_cleanup_pending()
|
|
cfg = _provider_config(service)
|
|
if not cfg.get("auth_url") or not cfg.get("token_url"):
|
|
raise RuntimeError(f"Provider '{service}' hat keine auth_url/token_url. "
|
|
f"In oauth_apps.json eintragen oder einen der "
|
|
f"vordefinierten Services nutzen ({', '.join(DEFAULT_PROVIDERS)}).")
|
|
cid, _ = _provider_credentials(service)
|
|
redirect_uri = _callback_url(service)
|
|
state = secrets.token_urlsafe(32)
|
|
use_scopes = scopes if scopes else cfg.get("scopes") or []
|
|
|
|
params = {
|
|
"client_id": cid,
|
|
"response_type": "code",
|
|
"redirect_uri": redirect_uri,
|
|
"state": state,
|
|
}
|
|
if use_scopes:
|
|
params["scope"] = " ".join(use_scopes)
|
|
params.update(cfg.get("extra_auth_params") or {})
|
|
if extra_params:
|
|
params.update(extra_params)
|
|
|
|
url = cfg["auth_url"] + "?" + urllib.parse.urlencode(params)
|
|
|
|
_PENDING[state] = {
|
|
"service": service,
|
|
"redirect_uri": redirect_uri,
|
|
"scopes": use_scopes,
|
|
"created_at": time.time(),
|
|
}
|
|
logger.info("[oauth] Authorize-URL fuer %s gebaut: state=%s redirect=%s",
|
|
service, state[:8] + "...", redirect_uri)
|
|
return {"url": url, "state": state, "redirect_uri": redirect_uri, "service": service}
|
|
|
|
|
|
# ── Token-Exchange ──────────────────────────────────────────
|
|
|
|
|
|
def _token_request(token_url: str, body_params: dict, cfg: dict,
|
|
client_id: str, client_secret: str) -> dict:
|
|
"""POST an provider /token endpoint. Returns parsed JSON oder wirft."""
|
|
data = urllib.parse.urlencode(body_params).encode("utf-8")
|
|
headers = {"Content-Type": "application/x-www-form-urlencoded"}
|
|
if cfg.get("accept_header"):
|
|
headers["Accept"] = cfg["accept_header"]
|
|
# Client-Auth: 'basic' (Header) oder 'body' (im Form-Body)
|
|
if cfg.get("client_auth") == "basic":
|
|
auth_str = f"{client_id}:{client_secret}"
|
|
b64 = base64.b64encode(auth_str.encode("utf-8")).decode("ascii")
|
|
headers["Authorization"] = f"Basic {b64}"
|
|
else:
|
|
# bereits im body_params drin (siehe Caller)
|
|
pass
|
|
req = urllib.request.Request(token_url, data=data, method="POST", headers=headers)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
raw = resp.read().decode("utf-8", "ignore")
|
|
try:
|
|
return json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
# GitHub default ist form-urlencoded — accept_header sollte
|
|
# JSON anfordern, aber falls's doch mal kommt:
|
|
parsed = urllib.parse.parse_qs(raw)
|
|
return {k: v[0] if isinstance(v, list) and v else v for k, v in parsed.items()}
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read().decode("utf-8", "ignore")[:500]
|
|
raise RuntimeError(f"Token-Request HTTP {e.code}: {body}") from e
|
|
|
|
|
|
def handle_callback(service: str, code: str, state: str,
|
|
error: Optional[str] = None,
|
|
error_description: Optional[str] = None) -> dict:
|
|
"""Verarbeitet einen OAuth-Callback. Validiert state, tauscht code gegen
|
|
Token, speichert. Returns {ok, service, message, ...}.
|
|
|
|
Wird von /internal/oauth-callback (HTTP, von aria-bridge) gerufen.
|
|
"""
|
|
_cleanup_pending()
|
|
|
|
if error:
|
|
# Provider hat User-Abbruch oder Fehler gemeldet
|
|
_PENDING.pop(state, None) if state else None
|
|
logger.warning("[oauth] Provider-Error %s/%s: %s — %s",
|
|
service, state[:8] + "..." if state else "?", error, error_description)
|
|
return {"ok": False, "service": service, "error": error,
|
|
"errorDescription": error_description}
|
|
|
|
pending = _PENDING.pop(state, None)
|
|
if not pending:
|
|
logger.warning("[oauth] Unknown state %s fuer %s — abgelaufen oder CSRF?", state[:8] + "...", service)
|
|
return {"ok": False, "service": service,
|
|
"error": "invalid_state",
|
|
"errorDescription": "Unbekannter oder abgelaufener state (Auth-Anfrage muss erst per oauth_authorize neu gestartet werden)."}
|
|
if pending.get("service") != service:
|
|
logger.warning("[oauth] state-Service-Mismatch: pending=%s vs callback=%s",
|
|
pending.get("service"), service)
|
|
return {"ok": False, "service": service,
|
|
"error": "service_mismatch",
|
|
"errorDescription": "state gehoert zu einem anderen Service."}
|
|
|
|
if not code:
|
|
return {"ok": False, "service": service, "error": "no_code"}
|
|
|
|
cfg = _provider_config(service)
|
|
try:
|
|
client_id, client_secret = _provider_credentials(service)
|
|
except RuntimeError as exc:
|
|
return {"ok": False, "service": service, "error": "no_credentials",
|
|
"errorDescription": str(exc)}
|
|
|
|
body = {
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": pending["redirect_uri"],
|
|
}
|
|
if cfg.get("client_auth") != "basic":
|
|
body["client_id"] = client_id
|
|
body["client_secret"] = client_secret
|
|
|
|
try:
|
|
token_data = _token_request(cfg["token_url"], body, cfg, client_id, client_secret)
|
|
except Exception as exc:
|
|
logger.exception("[oauth] Token-Exchange fehlgeschlagen fuer %s", service)
|
|
return {"ok": False, "service": service, "error": "exchange_failed",
|
|
"errorDescription": str(exc)[:200]}
|
|
|
|
access = token_data.get("access_token")
|
|
if not access:
|
|
return {"ok": False, "service": service, "error": "no_access_token",
|
|
"errorDescription": str(token_data)[:200]}
|
|
|
|
expires_in = int(token_data.get("expires_in") or 3600)
|
|
refresh = token_data.get("refresh_token") or ""
|
|
scope = token_data.get("scope") or " ".join(pending.get("scopes") or [])
|
|
token_type = token_data.get("token_type") or "Bearer"
|
|
|
|
record = {
|
|
"service": service,
|
|
"access_token": access,
|
|
"refresh_token": refresh,
|
|
"token_type": token_type,
|
|
"scope": scope,
|
|
"expires_at": int(time.time()) + expires_in,
|
|
"obtained_at": int(time.time()),
|
|
}
|
|
_persist_token(service, record)
|
|
logger.info("[oauth] %s authentifiziert — expires in %ds, refresh=%s",
|
|
service, expires_in, "ja" if refresh else "nein")
|
|
return {"ok": True, "service": service, "expiresIn": expires_in,
|
|
"hasRefresh": bool(refresh), "scope": scope}
|
|
|
|
|
|
# ── Token-Storage / Refresh / Revoke ─────────────────────────
|
|
|
|
|
|
def _persist_token(service: str, record: dict) -> None:
|
|
tokens = _load_json(TOKENS_FILE)
|
|
tokens[service] = record
|
|
_save_json(TOKENS_FILE, tokens)
|
|
|
|
|
|
def _load_token(service: str) -> Optional[dict]:
|
|
return _load_json(TOKENS_FILE).get(service)
|
|
|
|
|
|
def get_token(service: str, refresh_threshold_sec: int = 60) -> dict:
|
|
"""Holt das aktuelle access_token fuer einen Service. Refresht automatisch
|
|
wenn weniger als refresh_threshold_sec Restzeit. Returns das ganze
|
|
record-dict — Caller nimmt sich access_token raus.
|
|
|
|
Wirft wenn nicht authentifiziert oder Refresh fehlschlaegt — Tool-Aufrufer
|
|
soll dann oauth_authorize anbieten."""
|
|
record = _load_token(service)
|
|
if not record:
|
|
raise RuntimeError(f"Kein Token fuer '{service}' gespeichert. Erst per "
|
|
f"oauth_authorize authentifizieren.")
|
|
exp = int(record.get("expires_at") or 0)
|
|
remaining = exp - int(time.time())
|
|
if remaining > refresh_threshold_sec:
|
|
return record
|
|
# Refresh noetig
|
|
refresh_tok = (record.get("refresh_token") or "").strip()
|
|
if not refresh_tok:
|
|
raise RuntimeError(f"Token fuer '{service}' abgelaufen und kein refresh_token "
|
|
f"vorhanden — bitte neu autorisieren mit oauth_authorize.")
|
|
cfg = _provider_config(service)
|
|
client_id, client_secret = _provider_credentials(service)
|
|
body = {
|
|
"grant_type": "refresh_token",
|
|
"refresh_token": refresh_tok,
|
|
}
|
|
if cfg.get("client_auth") != "basic":
|
|
body["client_id"] = client_id
|
|
body["client_secret"] = client_secret
|
|
try:
|
|
new_data = _token_request(cfg["token_url"], body, cfg, client_id, client_secret)
|
|
except Exception as exc:
|
|
raise RuntimeError(f"Token-Refresh fuer '{service}' fehlgeschlagen: {exc}") from exc
|
|
|
|
new_access = new_data.get("access_token")
|
|
if not new_access:
|
|
raise RuntimeError(f"Refresh-Antwort ohne access_token: {new_data}")
|
|
expires_in = int(new_data.get("expires_in") or 3600)
|
|
# refresh_token kann (manche Provider) bei jedem Refresh rotieren
|
|
new_refresh = (new_data.get("refresh_token") or refresh_tok).strip()
|
|
record.update({
|
|
"access_token": new_access,
|
|
"refresh_token": new_refresh,
|
|
"expires_at": int(time.time()) + expires_in,
|
|
"obtained_at": int(time.time()),
|
|
})
|
|
if new_data.get("scope"):
|
|
record["scope"] = new_data["scope"]
|
|
_persist_token(service, record)
|
|
logger.info("[oauth] %s Token refreshed — neue Restzeit %ds", service, expires_in)
|
|
return record
|
|
|
|
|
|
def revoke(service: str) -> bool:
|
|
"""Entfernt das Token aus dem Storage (Best-Effort, kein Provider-Revoke-Call)."""
|
|
tokens = _load_json(TOKENS_FILE)
|
|
if service not in tokens:
|
|
return False
|
|
tokens.pop(service, None)
|
|
_save_json(TOKENS_FILE, tokens)
|
|
logger.info("[oauth] %s Token geloescht (lokal).", service)
|
|
return True
|
|
|
|
|
|
def allowed_hosts(service: str) -> list[str]:
|
|
"""Hosts, an die ein Token dieses Services geschickt werden darf.
|
|
|
|
SICHERHEITSKERN von `call()`: ohne diese Liste koennte ein Aufruf den
|
|
Zugangstoken an einen beliebigen Server schicken — die Anfrage-URL kommt
|
|
schliesslich vom Modell. Darum wird nur zugelassen, was zum Provider selbst
|
|
gehoert: die Hosts aus auth_url/token_url plus optional eingetragene
|
|
api_hosts. Subdomains desselben registrierbaren Namens zaehlen mit
|
|
(accounts.spotify.com -> api.spotify.com), fremde Domains nie.
|
|
"""
|
|
cfg = _provider_config(service)
|
|
hosts: set[str] = set()
|
|
for key in ("auth_url", "token_url"):
|
|
u = cfg.get(key) or ""
|
|
try:
|
|
h = urllib.parse.urlparse(u).hostname or ""
|
|
except Exception:
|
|
h = ""
|
|
if h:
|
|
hosts.add(h.lower())
|
|
for h in (cfg.get("api_hosts") or []):
|
|
h = str(h).strip().lower()
|
|
if h:
|
|
hosts.add(h)
|
|
return sorted(hosts)
|
|
|
|
|
|
def _host_allowed(host: str, allowed: list[str]) -> bool:
|
|
"""True, wenn host zu einem der erlaubten Hosts gehoert.
|
|
|
|
Verglichen wird auf den letzten ZWEI Labels (spotify.com). Ein reiner
|
|
endswith-Vergleich waere angreifbar: 'boesespotify.com' endet auch auf
|
|
'spotify.com'.
|
|
"""
|
|
h = (host or "").lower().strip(".")
|
|
if not h:
|
|
return False
|
|
|
|
def base(x: str) -> str:
|
|
parts = x.lower().strip(".").split(".")
|
|
return ".".join(parts[-2:]) if len(parts) >= 2 else x.lower()
|
|
|
|
hb = base(h)
|
|
return any(h == a or hb == base(a) for a in allowed)
|
|
|
|
|
|
def call(service: str, url: str, method: str = "GET",
|
|
body: Optional[str] = None, content_type: str = "application/json",
|
|
timeout: float = 20.0, max_bytes: int = 200_000) -> tuple[int, str]:
|
|
"""Fuehrt einen HTTP-Aufruf MIT dem Zugangstoken aus — ohne ihn herauszugeben.
|
|
|
|
Das ist der Ersatz fuer "gib mir den Token, ich rufe selbst": der Token
|
|
verlaesst den Brain-Prozess nicht und kann damit auch nicht in einem
|
|
Gespraechsprotokoll landen.
|
|
|
|
Returns (status_code, text). Wirft ValueError bei unerlaubter URL.
|
|
"""
|
|
svc = (service or "").strip()
|
|
if not svc:
|
|
raise ValueError("service ist Pflicht")
|
|
u = (url or "").strip()
|
|
if not u.startswith("https://"):
|
|
# Kein http:// — der Token ginge im Klartext ueber die Leitung.
|
|
raise ValueError("url muss mit https:// beginnen")
|
|
host = urllib.parse.urlparse(u).hostname or ""
|
|
allowed = allowed_hosts(svc)
|
|
if not allowed:
|
|
raise ValueError(
|
|
f"Fuer '{svc}' sind keine Hosts hinterlegt — erst oauth_register_provider aufrufen.")
|
|
if not _host_allowed(host, allowed):
|
|
raise ValueError(
|
|
f"Host '{host}' gehoert nicht zu '{svc}'. Erlaubt: {', '.join(allowed)}")
|
|
|
|
m = (method or "GET").upper()
|
|
if m not in ("GET", "POST", "PUT", "PATCH", "DELETE"):
|
|
raise ValueError(f"Methode nicht erlaubt: {method}")
|
|
|
|
record = get_token(svc) # refresht bei Bedarf
|
|
token = record.get("access_token", "")
|
|
ttype = record.get("token_type") or "Bearer"
|
|
|
|
data = body.encode("utf-8") if body else None
|
|
req = urllib.request.Request(u, data=data, method=m)
|
|
req.add_header("Authorization", f"{ttype} {token}")
|
|
req.add_header("Accept", "application/json")
|
|
if data is not None:
|
|
req.add_header("Content-Type", content_type)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
raw = resp.read(max_bytes + 1)
|
|
status = resp.status
|
|
except urllib.error.HTTPError as exc:
|
|
raw = exc.read(max_bytes + 1) if hasattr(exc, "read") else b""
|
|
status = exc.code
|
|
truncated = len(raw) > max_bytes
|
|
text = raw[:max_bytes].decode("utf-8", "replace")
|
|
if truncated:
|
|
text += "\n…(gekuerzt)"
|
|
logger.info("[oauth] call %s %s %s -> %s (%d Byte)", svc, m, host, status, len(raw))
|
|
return status, text
|
|
|
|
|
|
def list_services() -> list[dict]:
|
|
"""Diagnostik: zeigt fuer jeden konfigurierten Service ob Token da ist
|
|
+ Ablaufzeit. Wird von Diagnostic genutzt."""
|
|
apps = _load_json(APPS_FILE)
|
|
tokens = _load_json(TOKENS_FILE)
|
|
out = []
|
|
services = set(apps.keys()) | set(tokens.keys()) | set(DEFAULT_PROVIDERS.keys())
|
|
now = int(time.time())
|
|
for s in sorted(services):
|
|
app = apps.get(s) or {}
|
|
tok = tokens.get(s) or {}
|
|
configured = bool(app.get("client_id") and app.get("client_secret"))
|
|
out.append({
|
|
"service": s,
|
|
"configured": configured,
|
|
"authenticated": bool(tok.get("access_token")),
|
|
"expiresAt": tok.get("expires_at"),
|
|
"expiresInSec": (tok.get("expires_at", 0) - now) if tok.get("expires_at") else None,
|
|
"hasRefresh": bool(tok.get("refresh_token")),
|
|
"scope": tok.get("scope", ""),
|
|
"isDefault": s in DEFAULT_PROVIDERS,
|
|
})
|
|
return out
|