c1881f58e6
Inclui console handoff Desk↔Console (Spec 019), melhorias DNS Viewer (037), OpenPanel/Nextcloud/VM116 deploy notes, contracts stack e sidebar actualizado. Co-authored-by: Cursor <cursoragent@cursor.com>
102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
"""Vault temporário — API Token Cloudflare BYO (Spec 037). Nunca logar o token."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
_BYO_DIR = Path("/var/lib/ligbox-wizard/onboarding_dns_byo_tokens")
|
|
_TTL_HOURS = 24 * 7
|
|
|
|
|
|
@dataclass
|
|
class ByoCloudflareSession:
|
|
domain: str
|
|
api_token: str
|
|
zone_id: str
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def _vault_path(session_id: str, domain: str) -> Path:
|
|
key = hashlib.sha256(f"{session_id}:{domain.lower()}".encode()).hexdigest()
|
|
return _BYO_DIR / f"{key}.json"
|
|
|
|
|
|
def _purge_expired() -> None:
|
|
if not _BYO_DIR.is_dir():
|
|
return
|
|
now = _now()
|
|
for path in _BYO_DIR.glob("*.json"):
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
expires = datetime.fromisoformat(data["expires_at"])
|
|
if expires.tzinfo is None:
|
|
expires = expires.replace(tzinfo=timezone.utc)
|
|
if now > expires:
|
|
path.unlink(missing_ok=True)
|
|
except (json.JSONDecodeError, KeyError, ValueError):
|
|
path.unlink(missing_ok=True)
|
|
|
|
|
|
def save_byo_token(session_id: str, domain: str, api_token: str, zone_id: str) -> None:
|
|
session_id = (session_id or "").strip()
|
|
domain = domain.lower().strip().rstrip(".")
|
|
if len(session_id) < 8:
|
|
raise ValueError("Sessão inválida")
|
|
if not api_token or not zone_id:
|
|
raise ValueError("Token ou zone_id ausente")
|
|
_BYO_DIR.mkdir(parents=True, exist_ok=True)
|
|
_purge_expired()
|
|
expires = _now() + timedelta(hours=_TTL_HOURS)
|
|
_vault_path(session_id, domain).write_text(
|
|
json.dumps(
|
|
{
|
|
"domain": domain,
|
|
"api_token": api_token,
|
|
"zone_id": zone_id,
|
|
"expires_at": expires.isoformat(),
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def load_byo_token(session_id: str, domain: str) -> ByoCloudflareSession | None:
|
|
session_id = (session_id or "").strip()
|
|
domain = domain.lower().strip().rstrip(".")
|
|
if len(session_id) < 8:
|
|
return None
|
|
_purge_expired()
|
|
path = _vault_path(session_id, domain)
|
|
if not path.is_file():
|
|
return None
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
expires = datetime.fromisoformat(data["expires_at"])
|
|
if expires.tzinfo is None:
|
|
expires = expires.replace(tzinfo=timezone.utc)
|
|
if _now() > expires:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
return ByoCloudflareSession(
|
|
domain=data["domain"],
|
|
api_token=data["api_token"],
|
|
zone_id=data["zone_id"],
|
|
)
|
|
except (json.JSONDecodeError, KeyError, ValueError):
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
|
|
|
|
def clear_byo_token(session_id: str, domain: str) -> None:
|
|
session_id = (session_id or "").strip()
|
|
domain = domain.lower().strip().rstrip(".")
|
|
if len(session_id) < 8:
|
|
return
|
|
_vault_path(session_id, domain).unlink(missing_ok=True)
|