Schlüssel-Proxy, Server-Anforderungen und gitleaks-Regel für API-Werk
- tools/ki-proxy: setzt API-Schlüssel aus der Windows-Anmeldeinformationsverwaltung ein, nur im Speicher, ohne Header-Protokoll; Tests gegen Platzhalter-Server - docs/server-anforderungen.md für den ITM-Entwickler - .gitleaks.toml erkennt zki_/zodl_/zocr_-Schlüssel - CLAUDE.md, Plan, Tech-Stack: Schlüssel nie in .env, in der App verschlüsselt in der DB Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Christoph
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
6d6f8f8f27
commit
ca09d23a78
@@ -0,0 +1,65 @@
|
||||
# Schlüssel-Proxy für die KI-API (nur Entwicklung)
|
||||
|
||||
Der Proxy setzt die API-Schlüssel von ITM in Anfragen an `https://api-werk.de` ein. Claude Code,
|
||||
Tests und die App in Sail rufen den Proxy ohne Schlüssel auf und sehen ihn nie. Die Schlüssel
|
||||
liegen in der Windows-Anmeldeinformationsverwaltung. Der Proxy liest sie beim Start und hält sie
|
||||
nur im Arbeitsspeicher; er protokolliert Methode, Pfad, Status und Dauer, aber keine Header.
|
||||
|
||||
In Staging und Produktion gibt es keinen Proxy: Dort stehen die Schlüssel verschlüsselt in der
|
||||
Datenbank (Verwaltung → Einstellungen).
|
||||
|
||||
## Einmalig: Schlüssel in den Windows-Tresor eintragen
|
||||
|
||||
Systemsteuerung → Anmeldeinformationsverwaltung → Windows-Anmeldeinformationen →
|
||||
„Generische Anmeldeinformationen hinzufügen“:
|
||||
|
||||
| Internet- oder Netzwerkadresse | Benutzername | Kennwort |
|
||||
|---|---|---|
|
||||
| `ki-bauin/itm-ki` | `api` | KI-Schlüssel (`zki_…`) |
|
||||
| `ki-bauin/dataloader` | `api` | OpenDataLoader-Schlüssel (`zodl_…`) |
|
||||
| `ki-bauin/ocr` | `api` | OCR-Schlüssel (`zocr_…`) |
|
||||
|
||||
Schlüssel nie in ein Terminal, eine Datei oder den Chat mit Claude kopieren.
|
||||
|
||||
## Starten (nur du, in einem eigenen Terminal)
|
||||
|
||||
```bash
|
||||
cd ~/code/ki-bauin
|
||||
python3 tools/ki-proxy/ki_proxy.py --check # zeigt nur „gefunden“ / „FEHLT“, keine Werte
|
||||
python3 tools/ki-proxy/ki_proxy.py # läuft, bis Strg+C
|
||||
```
|
||||
|
||||
Nach dem Ändern eines Schlüssels den Proxy neu starten. Claude Code startet den Proxy nie selbst;
|
||||
ein Hook in `~/.claude/settings.json` sperrt das zusätzlich.
|
||||
|
||||
## Adressen
|
||||
|
||||
| Von | Adresse |
|
||||
|---|---|
|
||||
| WSL (Claude Code, Skripte) | `http://127.0.0.1:8787` |
|
||||
| App in Sail (Container) | `http://host.docker.internal:8787` |
|
||||
|
||||
Pfade wie bei api-werk.de: `/v1/ki/…`, `/v1/dataloader/…`, `/v1/ocr/…`. Andere Pfade lehnt der
|
||||
Proxy ab. Über die Netzwerkkarte der WSL und aus dem LAN ist er nicht erreichbar.
|
||||
|
||||
```bash
|
||||
curl -s http://127.0.0.1:8787/v1/ki/models
|
||||
```
|
||||
|
||||
## Was der Proxy schützt und was nicht
|
||||
|
||||
- **Geschützt:** Der Schlüssel steht in keiner Datei, keiner Umgebungsvariablen und keinem
|
||||
Startbefehl. Den Speicher des Proxys können andere Prozesse nicht lesen (`ptrace_scope=1`).
|
||||
Ein Hook sperrt für Claude Code Windows-Programme (`powershell.exe`, `cmdkey.exe` usw.), das
|
||||
Auslesen fremder Prozesse und das Starten des Proxys.
|
||||
- **Nicht geschützt:** Das ist ein Schutz vor Versehen, kein Schutz vor Absicht. Jeder Prozess
|
||||
auf deinem Rechner kann den Proxy aufrufen, solange er läuft. Und der Proxy schützt die
|
||||
Schlüssel, nicht die Daten: Echte Kundendaten gehen trotzdem nicht an Claude.
|
||||
|
||||
## Tests
|
||||
|
||||
```bash
|
||||
python3 -m unittest discover -s tools/ki-proxy
|
||||
```
|
||||
|
||||
Die Tests laufen gegen einen Platzhalter-Server, ohne Tresor und ohne Netz.
|
||||
@@ -0,0 +1,300 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Schlüssel-Proxy für die KI-API von ITM (API-Werk) – nur für die Entwicklung.
|
||||
|
||||
Liest beim Start die API-Schlüssel aus der Windows-Anmeldeinformationsverwaltung,
|
||||
hält sie nur im Arbeitsspeicher und setzt sie in weitergeleitete Anfragen ein.
|
||||
Aufrufer (Claude Code, Tests, die App in Sail) schicken Anfragen ohne Schlüssel.
|
||||
|
||||
Der Proxy wird ausschließlich vom Entwickler in einem eigenen Terminal gestartet,
|
||||
nie von Claude Code. Er gibt Schlüssel nie aus und protokolliert keine Header.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import http.client
|
||||
import json
|
||||
import logging
|
||||
import shutil
|
||||
import ssl
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from typing import Callable, Optional
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
# Pfadpräfix → Eintrag in der Windows-Anmeldeinformationsverwaltung
|
||||
ROUTES: dict[str, str] = {
|
||||
"/v1/ki/": "ki-bauin/itm-ki",
|
||||
"/v1/dataloader/": "ki-bauin/dataloader",
|
||||
"/v1/ocr/": "ki-bauin/ocr",
|
||||
}
|
||||
|
||||
DEFAULT_UPSTREAM = "https://api-werk.de"
|
||||
DEFAULT_PORT = 8787
|
||||
DEFAULT_LISTEN = ("127.0.0.1", "172.17.0.1") # WSL selbst und die Docker-Brücke (Sail)
|
||||
MAX_BODY_BYTES = 60 * 1024 * 1024 # Dateigrenze der API ist 50 MB
|
||||
UPSTREAM_TIMEOUT = 300
|
||||
|
||||
HOP_BY_HOP = {
|
||||
"connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
|
||||
"te", "trailer", "trailers", "transfer-encoding", "upgrade",
|
||||
}
|
||||
|
||||
log = logging.getLogger("ki-proxy")
|
||||
|
||||
# Liest generische Einträge über die Windows-API CredRead. Ausgabe je Eintrag:
|
||||
# "<eintrag>\t<base64 des Werts in UTF-8>" oder "<eintrag>\t-", wenn er fehlt.
|
||||
_POWERSHELL_TEMPLATE = r"""
|
||||
$ErrorActionPreference = 'Stop'
|
||||
Add-Type -TypeDefinition @'
|
||||
using System;
|
||||
using System.Runtime.InteropServices;
|
||||
public static class KiBauinTresor {
|
||||
[StructLayout(LayoutKind.Sequential, CharSet = CharSet.Unicode)]
|
||||
private struct CREDENTIAL {
|
||||
public int Flags; public int Type; public string TargetName; public string Comment;
|
||||
public System.Runtime.InteropServices.ComTypes.FILETIME LastWritten;
|
||||
public int CredentialBlobSize; public IntPtr CredentialBlob; public int Persist;
|
||||
public int AttributeCount; public IntPtr Attributes; public string TargetAlias; public string UserName;
|
||||
}
|
||||
[DllImport("advapi32.dll", CharSet = CharSet.Unicode, SetLastError = true)]
|
||||
private static extern bool CredRead(string target, int type, int flags, out IntPtr credential);
|
||||
[DllImport("advapi32.dll")]
|
||||
private static extern void CredFree(IntPtr credential);
|
||||
public static string Read(string target) {
|
||||
IntPtr p;
|
||||
if (!CredRead(target, 1, 0, out p)) { return null; }
|
||||
try {
|
||||
CREDENTIAL c = (CREDENTIAL)Marshal.PtrToStructure(p, typeof(CREDENTIAL));
|
||||
if (c.CredentialBlobSize == 0) { return ""; }
|
||||
return Marshal.PtrToStringUni(c.CredentialBlob, c.CredentialBlobSize / 2);
|
||||
} finally { CredFree(p); }
|
||||
}
|
||||
}
|
||||
'@
|
||||
foreach ($t in @(__TARGETS__)) {
|
||||
$v = [KiBauinTresor]::Read($t)
|
||||
if ($v -eq $null) { [Console]::Out.WriteLine($t + "`t-") }
|
||||
else { [Console]::Out.WriteLine($t + "`t" + [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($v.Trim()))) }
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
def parse_tresor_output(output: str) -> dict[str, Optional[str]]:
|
||||
"""Wertet die Ausgabe des PowerShell-Skripts aus."""
|
||||
result: dict[str, Optional[str]] = {}
|
||||
for line in output.splitlines():
|
||||
line = line.strip()
|
||||
if not line or "\t" not in line:
|
||||
continue
|
||||
target, value = line.split("\t", 1)
|
||||
result[target] = None if value == "-" else base64.b64decode(value).decode("utf-8")
|
||||
return result
|
||||
|
||||
|
||||
def read_windows_credentials(targets: list[str]) -> dict[str, Optional[str]]:
|
||||
"""Liest die angegebenen Einträge aus der Windows-Anmeldeinformationsverwaltung."""
|
||||
powershell = shutil.which("powershell.exe")
|
||||
if powershell is None:
|
||||
raise RuntimeError("powershell.exe nicht gefunden – läuft das unter WSL mit Windows-Interop?")
|
||||
quoted = ", ".join("'" + t.replace("'", "''") + "'" for t in targets)
|
||||
script = _POWERSHELL_TEMPLATE.replace("__TARGETS__", quoted)
|
||||
encoded = base64.b64encode(script.encode("utf-16-le")).decode("ascii")
|
||||
completed = subprocess.run(
|
||||
[powershell, "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-EncodedCommand", encoded],
|
||||
capture_output=True,
|
||||
timeout=60,
|
||||
)
|
||||
if completed.returncode != 0:
|
||||
# Die Meldung von PowerShell enthält keine Schlüssel; sie kommt in der Windows-Codepage.
|
||||
detail = completed.stderr.decode("cp850", errors="replace").strip().splitlines()[:3]
|
||||
raise RuntimeError("Lesen aus dem Windows-Tresor fehlgeschlagen: " + " | ".join(detail))
|
||||
found = parse_tresor_output(completed.stdout.decode("ascii", errors="replace"))
|
||||
return {t: found.get(t) for t in targets}
|
||||
|
||||
|
||||
class _LimitedReader:
|
||||
"""Liest genau `remaining` Bytes aus dem Eingangsstrom, damit der Request-Body gestreamt wird."""
|
||||
|
||||
def __init__(self, stream, length: int):
|
||||
self._stream = stream
|
||||
self._remaining = length
|
||||
|
||||
def read(self, size: int = -1) -> bytes:
|
||||
if self._remaining <= 0:
|
||||
return b""
|
||||
if size < 0 or size > self._remaining:
|
||||
size = self._remaining
|
||||
chunk = self._stream.read(size)
|
||||
self._remaining -= len(chunk)
|
||||
return chunk
|
||||
|
||||
|
||||
class ProxyHandler(BaseHTTPRequestHandler):
|
||||
protocol_version = "HTTP/1.0" # Antworten enden mit dem Schließen der Verbindung; einfach für Streaming
|
||||
server_version = "ki-proxy"
|
||||
server: "KiProxyServer"
|
||||
|
||||
def do_GET(self) -> None:
|
||||
self._forward()
|
||||
|
||||
def do_POST(self) -> None:
|
||||
self._forward()
|
||||
|
||||
def do_PUT(self) -> None:
|
||||
self._forward()
|
||||
|
||||
def do_PATCH(self) -> None:
|
||||
self._forward()
|
||||
|
||||
def do_DELETE(self) -> None:
|
||||
self._forward()
|
||||
|
||||
def log_message(self, format: str, *args) -> None: # noqa: A002 – Standardausgabe abschalten
|
||||
return
|
||||
|
||||
def _send_json(self, status: int, payload: dict) -> None:
|
||||
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def _forward(self) -> None:
|
||||
started = time.monotonic()
|
||||
path_only = self.path.split("?", 1)[0]
|
||||
target = next((t for prefix, t in ROUTES.items() if path_only.startswith(prefix)), None)
|
||||
status = 0
|
||||
sent = 0
|
||||
try:
|
||||
if target is None:
|
||||
status = 404
|
||||
self._send_json(404, {"error": "Unbekannter Pfad. Erlaubt: " + ", ".join(ROUTES)})
|
||||
return
|
||||
key = self.server.keys.get(target)
|
||||
if not key:
|
||||
status = 503
|
||||
self._send_json(503, {"error": f"Schlüssel '{target}' fehlt im Windows-Tresor. Proxy nach dem Eintragen neu starten."})
|
||||
return
|
||||
if self.headers.get("Transfer-Encoding", "").lower() == "chunked":
|
||||
status = 411
|
||||
self._send_json(411, {"error": "Request-Body mit Content-Length senden, nicht chunked."})
|
||||
return
|
||||
length = int(self.headers.get("Content-Length") or 0)
|
||||
if length > MAX_BODY_BYTES:
|
||||
status = 413
|
||||
self._send_json(413, {"error": "Request zu groß für den Proxy."})
|
||||
return
|
||||
|
||||
headers = {
|
||||
name: value
|
||||
for name, value in self.headers.items()
|
||||
if name.lower() not in HOP_BY_HOP | {"authorization", "host", "content-length"}
|
||||
}
|
||||
headers["Authorization"] = f"Bearer {key}"
|
||||
if length or self.command in ("POST", "PUT", "PATCH"):
|
||||
headers["Content-Length"] = str(length)
|
||||
body = _LimitedReader(self.rfile, length) if length else None
|
||||
|
||||
conn = self.server.connect_upstream()
|
||||
try:
|
||||
conn.request(self.command, self.server.upstream_base_path + self.path, body=body, headers=headers)
|
||||
response = conn.getresponse()
|
||||
status = response.status
|
||||
self.send_response(response.status, response.reason)
|
||||
for name, value in response.getheaders():
|
||||
if name.lower() not in HOP_BY_HOP | {"server", "date"}:
|
||||
self.send_header(name, value)
|
||||
self.end_headers()
|
||||
while True:
|
||||
chunk = response.read1(65536)
|
||||
if not chunk:
|
||||
break
|
||||
self.wfile.write(chunk)
|
||||
self.wfile.flush()
|
||||
sent += len(chunk)
|
||||
finally:
|
||||
conn.close()
|
||||
except (OSError, http.client.HTTPException) as exc:
|
||||
if status == 0:
|
||||
status = 502
|
||||
try:
|
||||
self._send_json(502, {"error": f"KI-API nicht erreichbar ({type(exc).__name__})."})
|
||||
except OSError:
|
||||
pass
|
||||
finally:
|
||||
log.info("%s %s → %s, %d Bytes, %.0f ms", self.command, path_only, status or "-", sent, (time.monotonic() - started) * 1000)
|
||||
|
||||
|
||||
class KiProxyServer(ThreadingHTTPServer):
|
||||
daemon_threads = True
|
||||
|
||||
def __init__(self, address: tuple[str, int], keys: dict[str, Optional[str]], upstream: str):
|
||||
super().__init__(address, ProxyHandler)
|
||||
self.keys = keys
|
||||
parts = urlsplit(upstream)
|
||||
self.upstream_scheme = parts.scheme
|
||||
self.upstream_host = parts.hostname or ""
|
||||
self.upstream_port = parts.port or (443 if parts.scheme == "https" else 80)
|
||||
self.upstream_base_path = parts.path.rstrip("/")
|
||||
self._ssl_context = ssl.create_default_context() if parts.scheme == "https" else None
|
||||
|
||||
def connect_upstream(self) -> http.client.HTTPConnection:
|
||||
if self.upstream_scheme == "https":
|
||||
return http.client.HTTPSConnection(self.upstream_host, self.upstream_port, timeout=UPSTREAM_TIMEOUT, context=self._ssl_context)
|
||||
return http.client.HTTPConnection(self.upstream_host, self.upstream_port, timeout=UPSTREAM_TIMEOUT)
|
||||
|
||||
|
||||
def start_servers(hosts: list[str], port: int, keys: dict[str, Optional[str]], upstream: str) -> list[KiProxyServer]:
|
||||
servers = []
|
||||
for host in hosts:
|
||||
try:
|
||||
server = KiProxyServer((host, port), keys, upstream)
|
||||
except OSError as exc:
|
||||
log.warning("Kann nicht auf %s:%d lauschen (%s) – übersprungen.", host, port, exc.strerror)
|
||||
continue
|
||||
threading.Thread(target=server.serve_forever, daemon=True).start()
|
||||
servers.append(server)
|
||||
return servers
|
||||
|
||||
|
||||
def main(argv: Optional[list[str]] = None, reader: Callable[[list[str]], dict[str, Optional[str]]] = read_windows_credentials) -> int:
|
||||
parser = argparse.ArgumentParser(description="Schlüssel-Proxy für die KI-API von ITM (nur Entwicklung).")
|
||||
parser.add_argument("--port", type=int, default=DEFAULT_PORT)
|
||||
parser.add_argument("--listen", default=",".join(DEFAULT_LISTEN), help="Adressen, durch Komma getrennt")
|
||||
parser.add_argument("--upstream", default=DEFAULT_UPSTREAM)
|
||||
parser.add_argument("--check", action="store_true", help="Nur prüfen, welche Schlüssel im Tresor stehen, dann beenden")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s", datefmt="%H:%M:%S")
|
||||
|
||||
keys = reader(list(ROUTES.values()))
|
||||
for target in ROUTES.values():
|
||||
log.info("%-22s %s", target, "gefunden" if keys.get(target) else "FEHLT")
|
||||
if args.check:
|
||||
return 0 if all(keys.values()) else 1
|
||||
|
||||
hosts = [h.strip() for h in args.listen.split(",") if h.strip()]
|
||||
servers = start_servers(hosts, args.port, keys, args.upstream)
|
||||
if not servers:
|
||||
log.error("Keine Adresse verfügbar, Proxy beendet.")
|
||||
return 1
|
||||
log.info("Proxy läuft auf %s → %s (Strg+C beendet)", ", ".join(f"{s.server_address[0]}:{s.server_address[1]}" for s in servers), args.upstream)
|
||||
try:
|
||||
while True:
|
||||
time.sleep(3600)
|
||||
except KeyboardInterrupt:
|
||||
log.info("Proxy beendet.")
|
||||
finally:
|
||||
for server in servers:
|
||||
server.shutdown()
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,189 @@
|
||||
"""Tests für den Schlüssel-Proxy. Ausführen: python3 -m unittest discover -s tools/ki-proxy"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import http.client
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import unittest
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
|
||||
sys.path.insert(0, os.path.dirname(__file__))
|
||||
|
||||
import ki_proxy # noqa: E402
|
||||
|
||||
KEYS = {"ki-bauin/itm-ki": "test-ki-schluessel", "ki-bauin/dataloader": "test-dl-schluessel", "ki-bauin/ocr": None}
|
||||
|
||||
|
||||
class FakeUpstream(BaseHTTPRequestHandler):
|
||||
"""Platzhalter für api-werk.de: merkt sich die letzte Anfrage."""
|
||||
|
||||
protocol_version = "HTTP/1.1"
|
||||
last: dict = {}
|
||||
release_stream = threading.Event()
|
||||
|
||||
def log_message(self, format, *args): # noqa: A002
|
||||
return
|
||||
|
||||
def _record(self) -> bytes:
|
||||
length = int(self.headers.get("Content-Length") or 0)
|
||||
body = self.rfile.read(length) if length else b""
|
||||
FakeUpstream.last = {"method": self.command, "path": self.path, "headers": dict(self.headers.items()), "body": body}
|
||||
return body
|
||||
|
||||
def do_GET(self):
|
||||
self._record()
|
||||
if self.path.startswith("/v1/ki/stream"):
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "text/event-stream")
|
||||
self.send_header("Transfer-Encoding", "chunked")
|
||||
self.end_headers()
|
||||
self._chunk(b"data: erster\n\n")
|
||||
FakeUpstream.release_stream.wait(5)
|
||||
self._chunk(b"data: [DONE]\n\n")
|
||||
self.wfile.write(b"0\r\n\r\n")
|
||||
return
|
||||
status = 409 if self.path.startswith("/v1/dataloader/jobs/") else 200
|
||||
payload = json.dumps({"data": [{"id": "chat"}, {"id": "embed"}]}).encode()
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(payload)))
|
||||
self.end_headers()
|
||||
self.wfile.write(payload)
|
||||
|
||||
def do_POST(self):
|
||||
body = self._record()
|
||||
payload = json.dumps({"job_id": "abc", "bytes": len(body)}).encode()
|
||||
self.send_response(202)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(payload)))
|
||||
self.end_headers()
|
||||
self.wfile.write(payload)
|
||||
|
||||
def _chunk(self, data: bytes) -> None:
|
||||
self.wfile.write(f"{len(data):x}\r\n".encode() + data + b"\r\n")
|
||||
self.wfile.flush()
|
||||
|
||||
|
||||
class ProxyTest(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.upstream = ThreadingHTTPServer(("127.0.0.1", 0), FakeUpstream)
|
||||
threading.Thread(target=cls.upstream.serve_forever, daemon=True).start()
|
||||
upstream_url = f"http://127.0.0.1:{cls.upstream.server_address[1]}"
|
||||
cls.proxy = ki_proxy.start_servers(["127.0.0.1"], 0, dict(KEYS), upstream_url)[0]
|
||||
cls.port = cls.proxy.server_address[1]
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.proxy.shutdown()
|
||||
cls.upstream.shutdown()
|
||||
|
||||
def setUp(self):
|
||||
FakeUpstream.last = {}
|
||||
FakeUpstream.release_stream.clear()
|
||||
self.logs = []
|
||||
handler = logging.Handler()
|
||||
handler.emit = lambda record: self.logs.append(record.getMessage())
|
||||
ki_proxy.log.addHandler(handler)
|
||||
ki_proxy.log.setLevel(logging.INFO)
|
||||
self.addCleanup(ki_proxy.log.removeHandler, handler)
|
||||
|
||||
def request(self, method, path, body=None, headers=None):
|
||||
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
|
||||
conn.request(method, path, body=body, headers=headers or {})
|
||||
return conn, conn.getresponse()
|
||||
|
||||
def test_setzt_den_schluessel_der_route_ein(self):
|
||||
conn, response = self.request("GET", "/v1/ki/models")
|
||||
self.assertEqual(200, response.status)
|
||||
self.assertEqual(["chat", "embed"], [m["id"] for m in json.loads(response.read())["data"]])
|
||||
self.assertEqual("Bearer test-ki-schluessel", FakeUpstream.last["headers"]["Authorization"])
|
||||
conn.close()
|
||||
|
||||
def test_ueberschreibt_mitgeschickte_authorization(self):
|
||||
conn, response = self.request("GET", "/v1/ki/models", headers={"Authorization": "Bearer fremd"})
|
||||
response.read()
|
||||
self.assertEqual("Bearer test-ki-schluessel", FakeUpstream.last["headers"]["Authorization"])
|
||||
conn.close()
|
||||
|
||||
def test_upload_kommt_unveraendert_an_mit_eigenem_schluessel(self):
|
||||
body = os.urandom(300_000)
|
||||
headers = {"Content-Type": "multipart/form-data; boundary=x"}
|
||||
conn, response = self.request("POST", "/v1/dataloader/jobs?x=1", body=body, headers=headers)
|
||||
self.assertEqual(202, response.status)
|
||||
self.assertEqual(len(body), json.loads(response.read())["bytes"])
|
||||
self.assertEqual(body, FakeUpstream.last["body"])
|
||||
self.assertEqual("/v1/dataloader/jobs?x=1", FakeUpstream.last["path"])
|
||||
self.assertEqual("Bearer test-dl-schluessel", FakeUpstream.last["headers"]["Authorization"])
|
||||
conn.close()
|
||||
|
||||
def test_reicht_fehlerstatus_durch(self):
|
||||
conn, response = self.request("GET", "/v1/dataloader/jobs/abc/result?format=markdown")
|
||||
self.assertEqual(409, response.status)
|
||||
response.read()
|
||||
conn.close()
|
||||
|
||||
def test_fehlender_schluessel_ergibt_503_ohne_weiterleitung(self):
|
||||
conn, response = self.request("GET", "/v1/ocr/jobs/abc")
|
||||
self.assertEqual(503, response.status)
|
||||
self.assertIn("ki-bauin/ocr", json.loads(response.read())["error"])
|
||||
self.assertEqual({}, FakeUpstream.last)
|
||||
conn.close()
|
||||
|
||||
def test_unbekannter_pfad_ergibt_404(self):
|
||||
conn, response = self.request("GET", "/admin")
|
||||
self.assertEqual(404, response.status)
|
||||
response.read()
|
||||
self.assertEqual({}, FakeUpstream.last)
|
||||
conn.close()
|
||||
|
||||
def test_streaming_kommt_sofort_an(self):
|
||||
conn, response = self.request("GET", "/v1/ki/stream")
|
||||
self.assertEqual(200, response.status)
|
||||
first = response.read1(1024)
|
||||
self.assertIn(b"erster", first) # kommt an, bevor der Platzhalter den Rest freigibt
|
||||
self.assertNotIn(b"[DONE]", first)
|
||||
FakeUpstream.release_stream.set()
|
||||
rest = response.read()
|
||||
self.assertIn(b"[DONE]", rest)
|
||||
conn.close()
|
||||
|
||||
def test_protokoll_enthaelt_keine_schluessel(self):
|
||||
for path in ("/v1/ki/models", "/v1/ocr/x", "/unbekannt"):
|
||||
conn, response = self.request("GET", path, headers={"Authorization": "Bearer fremd"})
|
||||
response.read()
|
||||
conn.close()
|
||||
self.assertTrue(self.logs)
|
||||
for line in self.logs:
|
||||
for secret in ("test-ki-schluessel", "test-dl-schluessel", "fremd", "Bearer"):
|
||||
self.assertNotIn(secret, line)
|
||||
|
||||
|
||||
class TresorAusgabeTest(unittest.TestCase):
|
||||
def test_wertet_gefundene_und_fehlende_eintraege_aus(self):
|
||||
wert = base64.b64encode("zki_Äbc123".encode()).decode()
|
||||
output = f"ki-bauin/itm-ki\t{wert}\r\nki-bauin/ocr\t-\r\n"
|
||||
self.assertEqual({"ki-bauin/itm-ki": "zki_Äbc123", "ki-bauin/ocr": None}, ki_proxy.parse_tresor_output(output))
|
||||
|
||||
def test_check_meldet_fehlende_schluessel_ohne_werte_auszugeben(self):
|
||||
logs = []
|
||||
handler = logging.Handler()
|
||||
handler.emit = lambda record: logs.append(record.getMessage())
|
||||
ki_proxy.log.addHandler(handler)
|
||||
try:
|
||||
code = ki_proxy.main(["--check"], reader=lambda targets: dict(KEYS))
|
||||
finally:
|
||||
ki_proxy.log.removeHandler(handler)
|
||||
self.assertEqual(1, code)
|
||||
joined = "\n".join(logs)
|
||||
self.assertIn("FEHLT", joined)
|
||||
self.assertNotIn("test-ki-schluessel", joined)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user