From 4854b09a00cbd378fef7621d2618cc42f3e5ab80 Mon Sep 17 00:00:00 2001
From: dev6tools <tools@dev6hosting.com>
Date: Sun, 23 Aug 2026 15:37:58 -0400
Subject: [PATCH] feat: accept forge access tokens (Basic/Bearer/X-API-Key) for
 /fetch + API

---
 app.py | 72 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 70 insertions(+), 2 deletions(-)

diff --git a/app.py b/app.py
index 4847777..a3674af 100644
--- a/app.py
+++ b/app.py
@@ -1,11 +1,13 @@
 import asyncio
+import base64
 import json
 import os
 import secrets
 import time
+import urllib.request
 import uuid
 
-from fastapi import Depends, FastAPI, HTTPException, Request
+from fastapi import Depends, FastAPI, Header, HTTPException, Request
 from fastapi.responses import FileResponse, JSONResponse, Response
 from fastapi.security import HTTPBasic, HTTPBasicCredentials
 from pydantic import BaseModel
@@ -22,6 +24,49 @@ LOG_FILE = os.path.join(DATA_DIR, "logs.jsonl")
 PROXY = os.getenv("PROXY", "")
 ADMIN_USER = os.getenv("ADMIN_USER", "admin")
 ADMIN_PASS = os.getenv("ADMIN_PASS", "")  # empty = no auth (LAN mode)
+FORGEJO_API_URL = os.getenv("FORGEJO_API_URL", "").rstrip("/")  # validate tokens against forge
+_forge_cache: dict = {}  # token -> (ts, ok)
+
+
+def _forge_check(user: str, token: str) -> bool:
+    key = f"{user}:{token}"
+    now = time.time()
+    hit = _forge_cache.get(key)
+    if hit and now - hit[0] < 60:
+        return hit[1]
+    ok = False
+    try:
+        req = urllib.request.Request(FORGEJO_API_URL + "/api/v1/user")
+        cred = base64.b64encode(f"{user}:{token}".encode()).decode()
+        req.add_header("Authorization", "Basic " + cred)
+        with urllib.request.urlopen(req, timeout=5) as resp:
+            ok = resp.status == 200
+    except Exception:
+        ok = False
+    _forge_cache[key] = (now, ok)
+    return ok
+
+
+def _forge_whoami(token: str) -> bool:
+    """Validate a bare token (Bearer / X-API-Key): token used as basic-auth identity."""
+    key = f"token:{token}"
+    now = time.time()
+    hit = _forge_cache.get(key)
+    if hit and now - hit[0] < 60:
+        return hit[1]
+    ok = False
+    for password in ("x-oauth-basic", ""):
+        try:
+            req = urllib.request.Request(FORGEJO_API_URL + "/api/v1/user")
+            cred = base64.b64encode(f"{token}:{password}".encode()).decode()
+            req.add_header("Authorization", "Basic " + cred)
+            with urllib.request.urlopen(req, timeout=5) as resp:
+                ok = resp.status == 200
+                break
+        except Exception:
+            continue
+    _forge_cache[key] = (now, ok)
+    return ok
 CAPTCHA_WAIT_MS = int(os.getenv("CAPTCHA_WAIT_MS", "4000"))
 SOLVE_TIMEOUT_S = int(os.getenv("SOLVE_TIMEOUT_S", "900"))
 
@@ -41,7 +86,30 @@ _semaphore = asyncio.Semaphore(MAX_CONCURRENT)
 security = HTTPBasic(auto_error=False)
 
 
-def require_auth(creds: HTTPBasicCredentials | None = Depends(security)):
+def require_auth(
+    creds: HTTPBasicCredentials | None = Depends(security),
+    x_api_key: str | None = Header(default=None),
+    authorization: str | None = Header(default=None),
+):
+    if creds and secrets.compare_digest(creds.username, ADMIN_USER) and secrets.compare_digest(creds.password, ADMIN_PASS):
+        return creds  # local admin (WebUI / ops) always allowed
+    if FORGEJO_API_URL:
+        identity = None
+        if creds:
+            identity = (creds.username, creds.password)
+        elif x_api_key:
+            identity = ("token", x_api_key)
+        elif authorization and authorization.lower().startswith("bearer "):
+            identity = ("token", authorization.split(" ", 1)[1].strip())
+        if not identity:
+            raise HTTPException(401, "Missing forge credentials (Basic user:token, Bearer, or X-API-Key)")
+        if identity[0] == "token":
+            ok = _forge_whoami(identity[1])
+        else:
+            ok = _forge_check(identity[0], identity[1])
+        if not ok:
+            raise HTTPException(401, "Invalid forge token or insufficient scope")
+        return creds
     if not ADMIN_PASS:
         return None  # open mode
     if not creds or not secrets.compare_digest(creds.username, ADMIN_USER) or not secrets.compare_digest(creds.password, ADMIN_PASS):