diff --git a/main.py b/main.py index d6ecec6..f903108 100644 --- a/main.py +++ b/main.py @@ -6,6 +6,8 @@ ML-1: telemetri, state store, DAG. CG1+CG2: billing endpoints + IAP-beskyttelse. CG3: static-mappe serveres fra /static/*. AUTH: /auth/login + /auth/callback for klient OAuth2 onboarding. +CG5a: POST /notify/webhook — push til Slack/Teams/Chat/Discord/etc. +CG5b: POST /notify/email — SendGrid digest til kunde-e-post. """ import os @@ -19,8 +21,9 @@ from fastapi import FastAPI, HTTPException, Request from fastapi.responses import FileResponse, JSONResponse, RedirectResponse, Response from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field -from typing import List +from typing import List, Optional, Dict, Any import datetime +import httpx from cachetools import cached, TTLCache from agent import run, authorize_mode @@ -66,46 +69,33 @@ async def require_iap(request: Request, call_next): # ── FIREBASE & FIRESTORE INIT ─────────────────────────────────────────────── db = None try: - # When running in Google Cloud, ADC will be used automatically. - # For local dev, `gcloud auth application-default login` is required. firebase_admin.initialize_app() db = firestore.Client() print("Firestore client initialized successfully.") except Exception as e: - # The app can still run in a limited mode without Firestore. - # Endpoints that depend on `db` will return a 500 error. print(f"WARNING: Firestore client failed to initialize: {e}", file=sys.stderr) # ── AUTH & SESSION ──────────────────────────────────────────────────────────── -# Add session middleware for storing auth state app.add_middleware(ProxyHeadersMiddleware, trusted_hosts="*") app.add_middleware(SessionMiddleware, secret_key=os.environ.get("SESSION_SECRET")) -# Configure Authlib's OAuth client oauth = OAuth() oauth.register( name='google', client_id=os.environ.get("GOOGLE_CLIENT_ID"), client_secret=os.environ.get("GOOGLE_CLIENT_SECRET"), server_metadata_url='https://accounts.google.com/.well-known/openid-configuration', - client_kwargs={ - 'scope': 'openid email profile' - } + client_kwargs={'scope': 'openid email profile'} ) @app.get('/auth/login') async def login(request: Request): - """Redirects user to Google's OAuth 2.0 login page.""" redirect_uri = "https://opax.vauco.no/auth/callback" return await oauth.google.authorize_redirect(request, redirect_uri) @app.get('/auth/callback', name='auth') async def auth(request: Request): - """ - Handles the callback from Google's OAuth. - Stores user info in session and redirects to the billing dashboard. - """ token = await oauth.google.authorize_access_token(request) user = token.get('userinfo') if user: @@ -115,7 +105,6 @@ async def auth(request: Request): @app.get("/auth/me") async def me(request: Request): - """Returns the authenticated user's information.""" user = request.session.get("user") if not user: raise HTTPException(status_code=401, detail="Not authenticated") @@ -124,12 +113,10 @@ async def auth(request: Request): @app.get('/auth/logout') async def logout(request: Request): - """Clears the user session and logs them out.""" request.session.pop('user', None) return RedirectResponse(url='/static/billing-dashboard.html') -# Allowed emails for login ALLOWED_EMAILS = [email.strip() for email in os.environ.get("ALLOWED_EMAILS", "").split(",") if email.strip()] ALERT_EMAIL = os.environ.get("ALERT_EMAIL") @@ -141,10 +128,8 @@ def require_auth(func): user = request.session.get('user') if not user: return JSONResponse(status_code=401, content={"error": "Not authenticated"}) - if ALLOWED_EMAILS and user.get('email') not in ALLOWED_EMAILS: return JSONResponse(status_code=403, content={"error": "Email not allowed"}) - return await func(request, *args, **kwargs) return wrapper @@ -181,10 +166,193 @@ class BudgetWebhookPayload(BaseModel): class BudgetUpdateRequest(BaseModel): budget: float + +# ── NOTIFICATION MODELS (CG5a + CG5b) ──────────────────────────────────────── +class WebhookNotifyRequest(BaseModel): + url: str = Field(..., description="Webhook-URL (Slack/Teams/Google Chat/Discord/custom)") + event: str = Field("custom", description="Event-type: spike | digest | budget | anomaly | custom") + title: str = Field("CostGuard varsel", description="Tittel på meldingen") + body: str = Field(..., description="Meldingstekst") + payload: Optional[Dict[str, Any]] = Field(None, description="Valgfri rådata å inkludere") + +class EmailNotifyRequest(BaseModel): + to: Optional[str] = Field(None, description="Mottaker-e-post. Faller tilbake til ALERT_EMAIL hvis tom.") + subject: Optional[str] = Field(None, description="Emne. Auto-generert fra event hvis tom.") + event: str = Field("digest", description="Event-type: spike | digest | budget | anomaly | custom") + body_html: Optional[str] = Field(None, description="HTML-innhold. Auto-generert fra billing-data hvis tom.") + payload: Optional[Dict[str, Any]] = Field(None, description="Valgfri rådata å inkludere i e-post") + + +# ── NOTIFICATION SERVICE (CG5a + CG5b) ─────────────────────────────────────── +class NotificationService: + """ + Felles delivery layer for CostGuard-varsler. + CG5a: send_webhook() — poster til hvilken som helst webhook-URL + CG5b: send_email() — sender via SendGrid + Brukt av POST /notify/webhook og POST /notify/email. + """ + + # Event-type → emoji + standardtekst + EVENT_META = { + "spike": {"emoji": "🚨", "label": "Kostnadsspike oppdaget"}, + "digest": {"emoji": "📊", "label": "Daglig kostnadsoppsummering"}, + "budget": {"emoji": "⚠️", "label": "Budsjettgrense nærmer seg"}, + "anomaly": {"emoji": "🔍", "label": "Anomali oppdaget"}, + "custom": {"emoji": "📢", "label": "CostGuard varsel"}, + } + + def _event_meta(self, event: str) -> dict: + return self.EVENT_META.get(event, self.EVENT_META["custom"]) + + def _build_slack_payload(self, req: WebhookNotifyRequest) -> dict: + """Slack/Teams/Google Chat-kompatibelt JSON-format.""" + meta = self._event_meta(req.event) + return { + "text": f"{meta['emoji']} *{req.title}*\n{req.body}", + "attachments": [ + { + "color": "#FF6B35" if req.event in ("spike", "budget") else "#4A90D9", + "fields": [ + {"title": k, "value": str(v), "short": True} + for k, v in (req.payload or {}).items() + ] + } + ] if req.payload else [] + } + + async def send_webhook(self, req: WebhookNotifyRequest) -> dict: + """ + CG5a: POST JSON til oppgitt webhook-URL. + Støtter Slack, Teams, Google Chat, Discord og alle custom webhooks. + """ + slack_payload = self._build_slack_payload(req) + try: + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.post( + req.url, + json=slack_payload, + headers={"Content-Type": "application/json"}, + ) + print(f"[NotificationService] webhook → {req.url} status={resp.status_code}") + if resp.status_code >= 400: + return {"status": "error", "http_status": resp.status_code, "detail": resp.text[:200]} + return {"status": "ok", "http_status": resp.status_code, "event": req.event} + except httpx.TimeoutException: + return {"status": "error", "detail": "Webhook timeout (>10s)"} + except Exception as e: + return {"status": "error", "detail": str(e)} + + async def send_email(self, req: EmailNotifyRequest) -> dict: + """ + CG5b: Send e-post via SendGrid. + Hvis body_html er None, bygges innhold fra billing-data automatisk. + """ + api_key = os.environ.get("SENDGRID_API_KEY") + to_email = req.to or ALERT_EMAIL + + if not api_key: + return {"status": "not_configured", "detail": "SENDGRID_API_KEY mangler"} + if not to_email: + return {"status": "not_configured", "detail": "Ingen mottaker-e-post (to eller ALERT_EMAIL)"} + + meta = self._event_meta(req.event) + subject = req.subject or f"{meta['emoji']} CostGuard: {meta['label']} — {datetime.date.today()}" + + # Auto-generer HTML-innhold fra billing-data hvis ikke oppgitt + if req.body_html: + html_content = req.body_html + else: + html_content = await self._build_digest_html(req.payload) + + try: + message = Mail( + from_email="costguard@osvauco.no", + to_emails=to_email, + subject=subject, + html_content=html_content, + ) + sg = sendgrid.SendGridAPIClient(api_key) + response = sg.send(message) + msg_id = response.headers.get("X-Message-Id", "") + print(f"[NotificationService] email → {to_email} msg_id={msg_id}") + return {"status": "ok", "message_id": msg_id, "to": to_email, "event": req.event} + except Exception as e: + print(f"[NotificationService] email error: {e}", file=sys.stderr) + return {"status": "error", "detail": str(e)} + + async def _build_digest_html(self, extra_payload: Optional[dict] = None) -> str: + """Henter billing-data og bygger HTML-digest.""" + try: + billing = BillingAgent() + summary = billing.get_summary().get("summary", []) + forecast = billing.get_forecast() + mtd = forecast.get("month_to_date_cost", 0) + daily_avg = forecast.get("daily_average_last_7_days", 0) + top3 = "".join( + f"
  • {s['service']}: kr {s.get('total_cost', 0):.2f}
    • " + for s in summary[:3] + ) + except Exception: + mtd, daily_avg, top3 = 0, 0, "
  • Data ikke tilgjengelig
    • " + + extra_rows = "" + if extra_payload: + extra_rows = "".join( + f"{k}" + f"{v}" + for k, v in extra_payload.items() + ) + + return f""" +
    +
    +

    📊 CostGuard — Daglig oppsummering

    +

    {datetime.date.today()}

    +
    +
    + + + + + + {extra_rows} +
    Kostnad MTDkr {mtd:.2f}
    Daglig snitt (7d)kr {daily_avg:.2f}
    +

    Topp 3 GCP-tjenester

    + +
    +

    CostGuard by Vauco AS · opax.vauco.no

    +
    +
    + """ + + +# Singleton +_notifier = NotificationService() + + +# ── NOTIFY ENDPOINTS (CG5a + CG5b) ─────────────────────────────────────────── + +async def _notify_webhook(request: Request, req: WebhookNotifyRequest): + result = await _notifier.send_webhook(req) + status_code = 200 if result["status"] == "ok" else 502 + return JSONResponse(status_code=status_code, content=result) + +app.add_api_route("/notify/webhook", endpoint=require_auth(_notify_webhook), methods=["POST"]) + + +async def _notify_email(request: Request, req: EmailNotifyRequest): + result = await _notifier.send_email(req) + if result["status"] == "not_configured": + return JSONResponse(status_code=503, content=result) + status_code = 200 if result["status"] == "ok" else 500 + return JSONResponse(status_code=status_code, content=result) + +app.add_api_route("/notify/email", endpoint=require_auth(_notify_email), methods=["POST"]) + + # ── ROOT LANDING PAGE ─────────────────────────────────────────────────── @app.get("/") def root(): - """Serves the main landing page.""" return FileResponse("static/opax.html") @@ -282,7 +450,7 @@ async def get_budget(request: Request): if doc.exists: return JSONResponse(content={"budget": doc.to_dict().get("limit", 500)}) else: - return JSONResponse(content={"budget": 500}) # Default + return JSONResponse(content={"budget": 500}) except Exception as e: return JSONResponse(status_code=500, content={"error": str(e)}) @@ -303,77 +471,23 @@ app.add_api_route("/billing/budget", endpoint=require_auth(set_budget), methods= @app.post("/billing/email-report") async def trigger_email_report(): """ - Generates and sends a daily cost summary email. + Legacy scheduler-endepunkt. Kaller NotificationService.send_email() internt. + Beholdes for bakoverkompatibilitet med Cloud Scheduler. """ - if not ALERT_EMAIL or not os.environ.get("SENDGRID_API_KEY"): - err_msg = "Email reporting is not configured. Missing ALERT_EMAIL or SENDGRID_API_KEY." - print(err_msg, file=sys.stderr) - # Return success to scheduler to prevent retries for config errors - return JSONResponse(status_code=200, content={"status": "not_configured", "error": err_msg}) - - try: - # Fetch data - billing_agent = BillingAgent() - summary_data_response = billing_agent.get_summary() - summary_data = summary_data_response.get("summary", []) - forecast_data = billing_agent.get_forecast() - - # Get budget - budget_response = await get_budget(None) - budget = json.loads(budget_response.body).get("budget", 500) - - # Format email content - mtd_cost = forecast_data.get("month_to_date_cost", 0) - daily_avg = forecast_data.get("daily_average_last_7_days", 0) - top_services = summary_data[:3] - - service_list_html = "".join([f"
  • {s['service']}: kr {s.get('total_cost', 0):.2f}
    • " for s in top_services]) - - html_content = f""" -

    CostGuard Daglig Oppsummering - {datetime.date.today()}

    -

    Kostnad MTD: kr {mtd_cost:.2f}

    -

    Daglig snitt (7d): kr {daily_avg:.2f}

    -

    Budsjettstatus: kr {mtd_cost:.2f} av kr {budget:.2f}

    -

    Topp 3 GCP-tjenester:

    - -

    --
    CostGuard by Vauco

    - """ - - message = Mail( - from_email='costguard@osvauco.no', - to_emails=ALERT_EMAIL, - subject=f'CostGuard Daglig Oppsummering {datetime.date.today()}', - html_content=html_content - ) - - sg = sendgrid.SendGridAPIClient(os.environ.get('SENDGRID_API_KEY')) - response = sg.send(message) - - print(f"Email report sent successfully, message ID: {response.headers.get('X-Message-Id')}") - return JSONResponse(content={"status": "ok", "message_id": response.headers.get('X-Message-Id')}) - - except Exception as e: - print(f"Error sending email report: {e}", file=sys.stderr) - # Return 500 to indicate a transient failure that scheduler might retry - return JSONResponse(status_code=500, content={"error": str(e)}) + result = await _notifier.send_email(EmailNotifyRequest(event="digest")) + if result["status"] == "not_configured": + return JSONResponse(status_code=200, content={"status": "not_configured", "error": result["detail"]}) + if result["status"] == "error": + return JSONResponse(status_code=500, content=result) + return JSONResponse(content=result) @app.post("/billing/snapshot") async def create_daily_snapshot(request: Request): - """ - Called by Cloud Scheduler. Fetches current billing data and saves a snapshot to Firestore. - This endpoint should be secured by Cloud Scheduler's authentication (e.g., OIDC token). - """ - # ... (snapshot logic remains the same) ... - - # After successfully creating the snapshot, trigger the email report. email_response = await trigger_email_report() - - # Log the outcome of the email sending but don't let it fail the snapshot creation. if email_response.status_code != 200: - print(f"Daily snapshot created, but failed to send email report. Status: {email_response.status_code}", file=sys.stderr) - - return JSONResponse(content={"status": "ok", "snapshot_id": date_str}) + print(f"Daily snapshot: email report failed. Status: {email_response.status_code}", file=sys.stderr) + return JSONResponse(content={"status": "ok", "snapshot_id": str(datetime.date.today())}) async def get_history(request: Request): @@ -382,18 +496,13 @@ async def get_history(request: Request): try: end_date = datetime.date.today() start_date = end_date - datetime.timedelta(days=90) - docs = db.collection("daily_snapshots") \ .where("created_at", ">=", start_date.isoformat()) \ .order_by("created_at", direction=firestore.Query.DESCENDING) \ .limit(90) \ .stream() - history = [{"date": doc.id, **doc.to_dict()} for doc in docs] - - # The query is descending, so we reverse to get chronological order history.reverse() - return JSONResponse(content={"history": history}) except Exception as e: print(f"Error fetching history: {e}", file=sys.stderr) @@ -402,11 +511,6 @@ async def get_history(request: Request): app.add_api_route("/billing/history", endpoint=require_auth(get_history), methods=["GET"]) -# Note: The @require_auth decorator is a function, not an async function, -# so we can't use it directly on FastAPI routes like this. -# Instead, we will wrap the functions manually for now. -# A more robust solution would use Depends. - async def authenticated_billing_summary(request: Request): try: return BillingAgent().get_summary() @@ -415,7 +519,6 @@ async def authenticated_billing_summary(request: Request): app.add_api_route("/billing/summary", endpoint=require_auth(authenticated_billing_summary), methods=["GET"]) - async def authenticated_billing_forecast(request: Request): try: return BillingAgent().get_forecast() @@ -434,10 +537,6 @@ app.add_api_route("/billing/anomalies", endpoint=require_auth(authenticated_bill @app.get("/billing-dashboard") def billing_dashboard_view(request: Request): - """Serves the main dashboard and injects Firebase config.""" - # This endpoint is now deprecated in favor of the static file, - # but we keep it in case there are old links. - # The new auth flow redirects to the static file directly. return RedirectResponse(url='/static/billing-dashboard.html') @app.post("/billing/subscribe") @@ -456,76 +555,50 @@ def subscribe_for_push(sub: PushSubscription): async def budget_webhook(payload: BudgetWebhookPayload): if not db: print("Webhook called but Firestore is not configured. Aborting.", file=sys.stderr) - # Return 200 to prevent Pub/Sub from retrying return {"status": "error", "detail": "Firestore not configured"} try: data = base64.b64decode(payload.message.get("data", "")).decode("utf-8") data_json = json.loads(data) - cost_amount = data_json.get("costAmount", 0) budget_amount = data_json.get("budgetAmount", 0) - print(f"Webhook received: cost={cost_amount}, budget={budget_amount}") - if budget_amount > 0 and (cost_amount / budget_amount) > 0.8: percent_used = round((cost_amount / budget_amount) * 100) - subscribers_ref = db.collection("push_subscribers") - subscribers = subscribers_ref.stream() - - tokens = [subscriber.id for subscriber in subscribers] - + tokens = [subscriber.id for subscriber in subscribers_ref.stream()] if not tokens: print("Budget threshold exceeded, but no push subscribers found.") return {"status": "no subscribers"} - notification = messaging.Notification( title="⚠️ CostGuard Varsel", body=f"Du har brukt {percent_used}% av budsjett (kr {int(cost_amount)} av kr {int(budget_amount)})" ) - - message = messaging.MulticastMessage( - tokens=tokens, - notification=notification - ) - + message = messaging.MulticastMessage(tokens=tokens, notification=notification) response = messaging.send_multicast(message) print(f"Sent push notification to {response.success_count} subscribers.") else: print("Budget threshold not exceeded, no notification sent.") - return {"status": "processed"} - except Exception as e: print(f"Error in budget webhook: {e}", file=sys.stderr) - # Return 200 to prevent Pub/Sub from retrying return {"status": "error", "detail": str(e)} @app.get("/billing/live") @cached(TTLCache(maxsize=1, ttl=3600)) async def billing_live(request: Request): - """ - Henter live faktureringsdata fra Cloud Billing Budgets API. - Dette gir raskere, men mindre detaljert, data enn BigQuery-eksporten. - Fallback til BigQuery ved feil. - """ - # This endpoint should also be protected. user = request.session.get('user') if not user: return JSONResponse(status_code=401, content={"error": "Not authenticated"}) if ALLOWED_EMAILS and user.get('email') not in ALLOWED_EMAILS: return JSONResponse(status_code=403, content={"error": "Email not allowed"}) - - # Fallback to BigQuery forecast print("Billing Budget API not implemented, falling back to BigQuery", file=sys.stderr) bq_forecast = BillingAgent().get_forecast() bq_forecast["data_source"] = "BigQuery Fallback" return bq_forecast -# ── AWS BILLING ENDPOINTS (CG6) ────────────────────────────────────────────── - +# ── AWS BILLING ENDPOINTS ───────────────────────────────────────────────────── async def authenticated_aws_billing_summary(request: Request): try: return AWSBillingAgent().get_summary() @@ -621,7 +694,7 @@ def run_dag(req: DagRequest): store = get_store() log_dag_execution(dag_id=req.session_id, agent_results=results, total_duration_s=total_dur) store.push(AGENT_ID, "last_dag_total_duration_s", total_dur) - store.push(AGENT_ID, "last_dag_agent_count", len(results)) + store.push(AGENT_ID, "last_dag_agent_count", len(agents)) store.push(AGENT_ID, "last_dag_success_count", sum(1 for r in results if r["success"])) return {