251 lines
8.9 KiB
Python
251 lines
8.9 KiB
Python
"""
|
|
NetAdmin — DNS & Mail Manager Backend
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
import uvicorn
|
|
from contextlib import asynccontextmanager
|
|
from fastapi import FastAPI, Request, Depends
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
|
|
from config import cfg
|
|
from broker import check_connection
|
|
import audit
|
|
from contextvars import ContextVar
|
|
|
|
# ── Request context (IP) ─────────────────────────────────────────────────
|
|
_request_ip: ContextVar[str] = ContextVar("request_ip", default="")
|
|
|
|
|
|
def get_request_ip() -> str:
|
|
return _request_ip.get()
|
|
|
|
|
|
class IPMiddleware(BaseHTTPMiddleware):
|
|
async def dispatch(self, request: Request, call_next):
|
|
ip = request.headers.get("X-Forwarded-For", "").split(",")[0].strip()
|
|
if not ip:
|
|
ip = request.client.host if request.client else ""
|
|
token = _request_ip.set(ip)
|
|
try:
|
|
return await call_next(request)
|
|
finally:
|
|
_request_ip.reset(token)
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
# ──────────────────────────────────────────────
|
|
# Sync helpers (run in thread pool — sync_from_agent is blocking)
|
|
# ──────────────────────────────────────────────
|
|
|
|
def _do_sync(label: str):
|
|
"""Run a full sync from both agents. Called from executor to avoid blocking."""
|
|
# ── DNS sync ─────────────────────────────────────────────────────
|
|
try:
|
|
from routers.dns import sync_from_agent as dns_sync
|
|
r = dns_sync(trigger=label)
|
|
log.info(
|
|
f"[sync:{label}] DNS OK — "
|
|
f"{r['total_zones']} zone(s), "
|
|
f"{r['imported_zones']} importée(s), "
|
|
f"{r['imported_records']} enreg."
|
|
)
|
|
except Exception as e:
|
|
log.warning(f"[sync:{label}] DNS failed (agent may be offline): {e}")
|
|
|
|
# ── Mail sync ─────────────────────────────────────────────────────
|
|
try:
|
|
from routers.mail import sync_from_agent as mail_sync
|
|
r = mail_sync(trigger=label)
|
|
log.info(
|
|
f"[sync:{label}] Mail OK — "
|
|
f"{r['total_domains']} domaine(s), "
|
|
f"{r['accounts']} nouveau(x) compte(s)"
|
|
)
|
|
except Exception as e:
|
|
log.warning(f"[sync:{label}] Mail failed (agent may be offline): {e}")
|
|
|
|
|
|
async def _agent_ready_listener():
|
|
"""Listen for agent.ready messages and flush the pending queue."""
|
|
import json as _json
|
|
from broker import get_redis, flush_pending_for_service
|
|
try:
|
|
r = get_redis()
|
|
sub_kwargs = {
|
|
"host": cfg.redis.host, "port": cfg.redis.port,
|
|
"db": cfg.redis.db, "decode_responses": True,
|
|
}
|
|
if cfg.redis.password:
|
|
sub_kwargs["password"] = cfg.redis.password
|
|
tls = cfg.redis.tls
|
|
if tls.enabled:
|
|
sub_kwargs.update({
|
|
"ssl": True,
|
|
"ssl_certfile": tls.client_cert,
|
|
"ssl_keyfile": tls.client_key,
|
|
"ssl_ca_certs": tls.ca_cert,
|
|
"ssl_cert_reqs": "required",
|
|
})
|
|
import redis as _redis
|
|
sub_r = _redis.Redis(**sub_kwargs)
|
|
pubsub = sub_r.pubsub()
|
|
pubsub.subscribe("agent.ready")
|
|
log.info("[agent-listener] Subscribed to agent.ready")
|
|
loop = asyncio.get_running_loop()
|
|
while True:
|
|
msg = await loop.run_in_executor(None, lambda: pubsub.get_message(timeout=1.0))
|
|
if msg and msg["type"] == "message":
|
|
try:
|
|
data = _json.loads(msg["data"])
|
|
service = data.get("service")
|
|
if service in ("dns", "mail"):
|
|
log.info(f"[agent-listener] Agent '{service}' reconnected — flushing queue")
|
|
audit.record("system", "agent_reconnect",
|
|
target=f"agent:{service}", detail=data)
|
|
await loop.run_in_executor(None, flush_pending_for_service, service)
|
|
except Exception as e:
|
|
log.warning(f"[agent-listener] Error processing agent.ready: {e}")
|
|
await asyncio.sleep(0)
|
|
except Exception as e:
|
|
log.warning(f"[agent-listener] Listener error: {e}")
|
|
|
|
|
|
async def _sync_once(label: str):
|
|
"""Async wrapper: run blocking sync in thread pool."""
|
|
if not check_connection():
|
|
log.warning(f"[sync:{label}] Redis not reachable — skipped")
|
|
return
|
|
loop = asyncio.get_running_loop()
|
|
await loop.run_in_executor(None, _do_sync, label)
|
|
|
|
|
|
async def _polling_loop():
|
|
"""Background task: sync from MariaDB every cfg.sync.interval seconds."""
|
|
interval = cfg.sync.interval
|
|
if interval <= 0:
|
|
log.info("[polling] Disabled (sync.interval=0)")
|
|
return
|
|
|
|
log.info(f"[polling] Started — interval={interval}s")
|
|
while True:
|
|
await asyncio.sleep(interval)
|
|
await _sync_once("poll")
|
|
|
|
|
|
# ──────────────────────────────────────────────
|
|
# Lifespan: startup sync + launch polling task
|
|
# ──────────────────────────────────────────────
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
# 1. Initial sync on startup
|
|
await _sync_once("startup") # trigger="startup" passed to sync functions
|
|
|
|
# 2. Launch background polling task
|
|
poll_task = None
|
|
listener_task = None
|
|
if cfg.sync.interval > 0:
|
|
poll_task = asyncio.create_task(_polling_loop())
|
|
log.info(f"[startup] Background polling task started (every {cfg.sync.interval}s)")
|
|
else:
|
|
log.info("[startup] Background polling disabled (sync.interval=0)")
|
|
|
|
try:
|
|
listener_task = asyncio.create_task(_agent_ready_listener())
|
|
log.info("[startup] Agent-ready listener started")
|
|
except Exception as e:
|
|
log.warning(f"[startup] Could not start agent-ready listener: {e}")
|
|
|
|
yield # ← application runs here
|
|
|
|
# 3. Shutdown: cancel tasks cleanly
|
|
for task in filter(None, [poll_task, listener_task]):
|
|
task.cancel()
|
|
try:
|
|
await task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
log.info("[shutdown] Background tasks stopped")
|
|
|
|
|
|
# ──────────────────────────────────────────────
|
|
# App
|
|
# ──────────────────────────────────────────────
|
|
|
|
app = FastAPI(
|
|
title="NetAdmin API",
|
|
version="2.0.0",
|
|
lifespan=lifespan,
|
|
docs_url=None, # Disable Swagger UI — not exposed to end users
|
|
redoc_url=None, # Disable ReDoc
|
|
openapi_url=None, # Disable OpenAPI schema endpoint
|
|
)
|
|
|
|
app.add_middleware(IPMiddleware)
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=cfg.server.cors_origins,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
from routers import auth, dns, mail # noqa: E402
|
|
from routers.auth import get_current_user, require_global_admin # noqa: E402
|
|
from db import get_db # noqa: E402
|
|
app.include_router(auth.router, prefix="/auth", tags=["Auth"])
|
|
app.include_router(dns.router, prefix="/dns", tags=["DNS"])
|
|
app.include_router(mail.router, prefix="/mail", tags=["Mail"])
|
|
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {
|
|
"status": "ok",
|
|
"redis": check_connection(),
|
|
"sync_interval": cfg.sync.interval,
|
|
}
|
|
|
|
|
|
@app.get("/sync/history")
|
|
def get_global_sync_history(
|
|
service: str | None = None,
|
|
trigger: str | None = None,
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
user: dict = Depends(require_global_admin),
|
|
):
|
|
"""Combined sync history for all services — global admin only."""
|
|
return {
|
|
"total": audit.count_sync_history(service),
|
|
"entries": audit.query_sync_history(service, trigger, limit, offset),
|
|
}
|
|
|
|
|
|
@app.get("/audit")
|
|
def get_audit_log(
|
|
service: str | None = None,
|
|
username: str | None = None,
|
|
action: str | None = None,
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
user: dict = Depends(require_global_admin),
|
|
):
|
|
"""Audit log — global admin only."""
|
|
return {
|
|
"total": audit.count(service, username, action),
|
|
"entries": audit.query(service, username, action, limit, offset),
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
uvicorn.run(
|
|
"main:app",
|
|
host=cfg.server.host,
|
|
port=cfg.server.port,
|
|
reload=False,
|
|
)
|