Compare commits

..
1 Commits
Author SHA1 Message Date
Julian Boldt 218a1f14ac Release v1.18.0
Release / verify (push) Canceled after 0s
CI / checks (push) Canceled after 0s
CI / frontend (push) Canceled after 0s
CI / backend (push) Canceled after 0s
Release / test (push) Canceled after 0s
Release / publish (backend) (push) Canceled after 0s
Release / publish (frontend) (push) Canceled after 0s
Release / publish (proxy) (push) Canceled after 0s
Release / publish (updater) (push) Canceled after 0s
Automatisierung und Sicherheitslage. Ende-zu-Ende gegen einen Test-Stack und drei Test-Firewalls geprüft,
siehe Handbuch „Sicherheitslage“ und „Automatisierung“.

- **API-Tokens** (System › API-Tokens): persönliche Tokens für Skripte, CI, Prometheus und KI-Assistenten,
  Bereiche *Lesen*, *Ausrollen*, *Admin*. Nur als Hash gespeichert, einmal angezeigt, widerrufbar, mit
  Ablaufdatum; Anlegen verlangt das Passwort. Ein Token überschreitet nie die Rechte seines Besitzers.
  Tokens verwalten und das eigene Konto geht nur in einer normalen Anmeldung. Geheimnisse (Roh-Backups,
  Benachrichtigungseinstellungen) liest nur der Bereich *Admin*. Das Audit nennt das Token. Neue Tabelle
  `api_tokens` (Migration `0008`).
- **Prometheus-Metriken** unter `GET /api/metrics` (Host erreichbar, Last, Speicher, Platte, CARP-Rolle,
  offene Aktionen und Probleme, Sicherheitswert), nur aus der Datenbank, ohne Anfragen an die Firewalls.
- **MCP-Endpunkt** `POST /api/mcp` (nur lesend) für KI-Assistenten: Flottenübersicht, Hosts, Auslastung, CARP-Ereignisse,
  Audit, Suche und ein Leser für jeden GET-Endpunkt, mit den Rechten des Tokens.
- **OpenAPI-Schema** jetzt unter `/api/openapi.json` (bisher unter `/openapi.json`, vom Proxy nicht
  erreichbar); die Swagger-/ReDoc-Seiten entfallen (ihre CDN-Skripte verbietet die CSP).
- **Sicherheitslage** (Monitoring › Sicherheitslage): Härtungs- und Regel-Hygiene-Prüfungen je Firewall mit Wert
  0–100: Allow-all am WAN, HTTPS, SSH, SNMP, verdeckte/doppelte/unbeschriftete Regeln, Zeitserver, externe
  Logs, Firmware, Backup-Alter. Liest die Konfiguration nur lesend, speichert sie nicht, alle 6 Stunden und
  auf Knopfdruck. Neue Tabelle `compliance_results` (Migration `0009`).
- Behoben: Änderungen am Benutzer in einer Anfrage (z. B. 2FA aktivieren) gingen verloren, wenn die
  Middleware den Benutzer aus einer anderen Datenbank-Sitzung lieferte; er wird jetzt in der Sitzung der
  Anfrage neu geladen.
2026-10-19 23:54:02 +02:00
37 changed files with 2898 additions and 45 deletions

No files matched your search

+26 -1
View File
@@ -29,7 +29,32 @@ in der App auf der Login-Seite, im Footer jeder Seite, auf `/about` und in
den Einstellungen.
Nächste Codenamen (der Reihe nach, wenn die jeweils nächste MINOR-Version
ansteht): Merlin …
ansteht): Lanner, Merlin …
## [1.18.0] "Kestrel" — 2026-10-19
Automatisierung und Sicherheitslage. Ende-zu-Ende gegen einen Test-Stack und drei Test-Firewalls geprüft,
siehe Handbuch „Sicherheitslage“ und „Automatisierung“.
- **API-Tokens** (System › API-Tokens): persönliche Tokens für Skripte, CI, Prometheus und KI-Assistenten,
Bereiche *Lesen*, *Ausrollen*, *Admin*. Nur als Hash gespeichert, einmal angezeigt, widerrufbar, mit
Ablaufdatum; Anlegen verlangt das Passwort. Ein Token überschreitet nie die Rechte seines Besitzers.
Tokens verwalten und das eigene Konto geht nur in einer normalen Anmeldung. Geheimnisse (Roh-Backups,
Benachrichtigungseinstellungen) liest nur der Bereich *Admin*. Das Audit nennt das Token. Neue Tabelle
`api_tokens` (Migration `0008`).
- **Prometheus-Metriken** unter `GET /api/metrics` (Host erreichbar, Last, Speicher, Platte, CARP-Rolle,
offene Aktionen und Probleme, Sicherheitswert), nur aus der Datenbank, ohne Anfragen an die Firewalls.
- **MCP-Endpunkt** `POST /api/mcp` (nur lesend) für KI-Assistenten: Flottenübersicht, Hosts, Auslastung, CARP-Ereignisse,
Audit, Suche und ein Leser für jeden GET-Endpunkt, mit den Rechten des Tokens.
- **OpenAPI-Schema** jetzt unter `/api/openapi.json` (bisher unter `/openapi.json`, vom Proxy nicht
erreichbar); die Swagger-/ReDoc-Seiten entfallen (ihre CDN-Skripte verbietet die CSP).
- **Sicherheitslage** (Monitoring › Sicherheitslage): Härtungs- und Regel-Hygiene-Prüfungen je Firewall mit Wert
0–100: Allow-all am WAN, HTTPS, SSH, SNMP, verdeckte/doppelte/unbeschriftete Regeln, Zeitserver, externe
Logs, Firmware, Backup-Alter. Liest die Konfiguration nur lesend, speichert sie nicht, alle 6 Stunden und
auf Knopfdruck. Neue Tabelle `compliance_results` (Migration `0009`).
- Behoben: Änderungen am Benutzer in einer Anfrage (z. B. 2FA aktivieren) gingen verloren, wenn die
Middleware den Benutzer aus einer anderen Datenbank-Sitzung lieferte; er wird jetzt in der Sitzung der
Anfrage neu geladen.
## [1.17.0] "Imperial" — 2026-10-19
+4 -2
View File
@@ -7,7 +7,7 @@ Zentrale Verwaltung mehrerer [OPNsense](https://opnsense.org)-Firewalls: Regeln,
![Zentrale Firewall-Regeln](docs/screenshots/readme-rules-de.webp)
> **Stand:** `1.17.0` ist die aktuelle stabile Version. Entwickelt und getestet wurde gegen echte OPNsense-Instanzen; eine frische Installation aus dem Release-Bundle sowie Updates und Rollbacks sind durchgängig geprüft. Ecken und Kanten bitte melden.
> **Stand:** `1.18.0` ist die aktuelle stabile Version. Entwickelt und getestet wurde gegen echte OPNsense-Instanzen; eine frische Installation aus dem Release-Bundle sowie Updates und Rollbacks sind durchgängig geprüft. Ecken und Kanten bitte melden.
## Funktionen
@@ -22,6 +22,8 @@ Zentrale Verwaltung mehrerer [OPNsense](https://opnsense.org)-Firewalls: Regeln,
- **Update-Center** — Firmware-Updates für die Flotte, mit Blick auf HA-Reihenfolge und Wartungsmodus.
- **Einrichtungsassistent** — geführter Erststart nach der Installation: eigenes Passwort, Zwei-Faktor und die erste Firewall verbinden.
- **Selbst-Update** — OPNexus aktualisiert sich auf Wunsch aus der eigenen Oberfläche, mit Fortschritt und automatischem Rollback (optional).
- **Automatisierung** — persönliche API-Tokens (Lesen / Ausrollen / Admin), Prometheus-Metriken, ein nur lesender MCP-Endpunkt für KI-Assistenten und ein OpenAPI-Schema.
- **Sicherheitslage** — ein Wert von 0 bis 100 je Firewall aus Härtungs- und Regel-Hygiene-Prüfungen (offene WAN-Regeln, schwaches SSH/SNMP, verdeckte Regeln, alte Backups …), nur lesend.
- **Zugang** — HTTPS mit gepinnten Firewall-Zertifikaten, optionale Zwei-Faktor-Anmeldung, Rollen (Admin / Nur-Lesen), Tastaturbedienung, Dark Mode, Oberfläche auf Deutsch und Englisch.
Alle Screenshots zeigen fiktive Beispieldaten.
@@ -58,7 +60,7 @@ Für den Zugriff von anderen Rechnern und den Produktivbetrieb mit TLS (mitgelie
```bash
OPNEXUS_IMAGE_REPO=git.opnexus.dev/opnexus
OPNEXUS_VERSION=1.17.0
OPNEXUS_VERSION=1.18.0
```
```bash
+4 -2
View File
@@ -7,7 +7,7 @@ Central management for multiple [OPNsense](https://opnsense.org) firewalls: rule
![Central firewall rules](docs/screenshots/readme-rules-en.webp)
> **Status:** `1.17.0` is the current stable release. It has been developed and tested against real OPNsense instances; a clean install from the release bundle, updates and rollbacks have been verified end to end. Please report rough edges.
> **Status:** `1.18.0` is the current stable release. It has been developed and tested against real OPNsense instances; a clean install from the release bundle, updates and rollbacks have been verified end to end. Please report rough edges.
## What it does
@@ -22,6 +22,8 @@ Central management for multiple [OPNsense](https://opnsense.org) firewalls: rule
- **Update center** — firmware updates for the fleet, aware of HA order and maintenance mode.
- **Setup assistant** — a guided first run after installation: your own password, two-factor, and connecting the first firewall.
- **Self-update** — OPNexus can update itself from its web interface, with progress and automatic rollback (opt-in).
- **Automation** — personal API tokens (read / deploy / admin), Prometheus metrics, a read-only MCP endpoint for AI assistants and an OpenAPI schema.
- **Security posture** — a 0–100 score per firewall from hardening and rule-hygiene checks (open WAN rules, weak SSH/SNMP, shadowed rules, stale backups …), read-only.
- **Access** — HTTPS with pinned firewall certificates, optional two-factor sign-in, roles (admin / read-only), keyboard operation, dark mode, German and English UI.
![Cluster view](docs/screenshots/readme-cluster-en.webp)
@@ -60,7 +62,7 @@ For access from other machines and for production use with TLS (bundled Caddy pr
```bash
OPNEXUS_IMAGE_REPO=git.opnexus.dev/opnexus
OPNEXUS_VERSION=1.17.0
OPNEXUS_VERSION=1.18.0
```
```bash
+28 -4
View File
@@ -8,7 +8,8 @@ from starlette.responses import JSONResponse
from app.database import SessionLocal
from app.errors import ErrorCode
from app.routers.auth import COOKIE_NAME
from app.services import auth_service
from app.routers.auth import client_ip
from app.services import api_token_service, auth_service
_PUBLIC_PATHS = {
"/api/auth/login",
@@ -35,6 +36,8 @@ _PUBLIC_PATHS = {
_READ_ONLY_ALLOWED_WRITES = {
"/api/auth/change-password",
"/api/settings",
# Read-only JSON-RPC tools (see routers/mcp.py).
"/api/mcp",
# Self-scoped: a user managing their own second factor.
"/api/auth/2fa/setup",
"/api/auth/2fa/enable",
@@ -48,6 +51,7 @@ _READ_ONLY_ALLOWED_WRITES = {
_SECURITY_KINDS = {
"ipsec": "ipsec_tunnel", "openvpn": "openvpn_instance", "wireguard": "wireguard_tunnel",
"certificates": "certificate", "acme": "acme", "notifications": "notification", "auth": "auth",
"tokens": "api_token",
}
_COLLECTIONS = {"tunnels", "instances", "managed", "accounts", "certificates", "settings"}
@@ -86,10 +90,26 @@ class AuthMiddleware(BaseHTTPMiddleware):
if request.url.path in _PUBLIC_PATHS:
return await call_next(request)
api_token = None
bearer = api_token_service.bearer_from_header(request.headers.get("authorization", ""))
token = request.cookies.get(COOKIE_NAME, "")
if token:
if bearer is not None:
# An API token wins over a cookie; a bad one never falls back to it.
async with SessionLocal() as db:
resolved = await api_token_service.resolve(db, bearer, client_ip(request))
user, api_token = resolved if resolved is not None else (None, None)
token = bearer
elif token:
async with SessionLocal() as db:
user = await auth_service.resolve_session(db, token)
if token:
if user is not None and api_token is not None and api_token_service.denies(
api_token.scope, request.method, request.url.path
):
return JSONResponse(
status_code=403,
content={"detail": {"code": ErrorCode.API_TOKEN_SCOPE_FORBIDDEN.value, "params": {"scope": api_token.scope}}},
)
if user is not None:
parts = request.url.path.strip("/").split("/")
kinds = {"rules": "rule", "aliases": "alias", "port-forward": "port_forward", "one-to-one": "one_to_one", "outbound-nat": "outbound_nat", "hosts": "host", "users": "user", "system": "system"}
@@ -109,10 +129,12 @@ class AuthMiddleware(BaseHTTPMiddleware):
user.role == "read_only"
and request.method not in ("GET", "HEAD", "OPTIONS")
and request.url.path not in _READ_ONLY_ALLOWED_WRITES
# Own API tokens (the router limits a read_only user to read scope).
and not request.url.path.startswith("/api/tokens")
):
if tracked:
# Refused attempts at security-relevant writes stay attributable.
ctx = audit_service.context.set({"actor": user.username, "id": str(uuid.uuid4()), "action": event["action"]})
ctx = audit_service.context.set({"actor": user.username if api_token is None else f"{user.username} (API: {api_token.name})"[:150], "id": str(uuid.uuid4()), "action": event["action"]})
try:
await audit_service.persist(**event, status="denied", message="Schreibzugriff für Nur-Lese-Konto verweigert")
finally:
@@ -122,7 +144,9 @@ class AuthMiddleware(BaseHTTPMiddleware):
content={"detail": {"code": ErrorCode.READ_ONLY_ACCESS.value, "params": {}}},
)
request.state.user = user
token = audit_service.context.set({"actor": user.username, "id": str(uuid.uuid4()), "action": event["action"]})
request.state.api_token = api_token
actor = user.username if api_token is None else f"{user.username} (API: {api_token.name})"[:150]
token = audit_service.context.set({"actor": actor, "id": str(uuid.uuid4()), "action": event["action"]})
try:
if tracked:
await audit_service.persist(**event, status="requested")
+4
View File
@@ -154,6 +154,10 @@ class ErrorCode(str, Enum):
SERVICE_NOT_FOUND = "SERVICE_NOT_FOUND"
SERVICE_LOCKED = "SERVICE_LOCKED"
READ_ONLY_ACCESS = "READ_ONLY_ACCESS"
API_TOKEN_LIMIT_REACHED = "API_TOKEN_LIMIT_REACHED"
API_TOKEN_NOT_FOUND = "API_TOKEN_NOT_FOUND"
API_TOKEN_SCOPE_FORBIDDEN = "API_TOKEN_SCOPE_FORBIDDEN"
API_TOKEN_SCOPE_TOO_HIGH = "API_TOKEN_SCOPE_TOO_HIGH"
def api_error(status_code: int, code: ErrorCode, **params: Any) -> HTTPException:
+58 -3
View File
@@ -5,6 +5,7 @@ from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.openapi.utils import get_openapi
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from sqlalchemy.exc import DBAPIError
@@ -17,7 +18,7 @@ from app.security_middleware import OriginCheckMiddleware, SecurityHeadersMiddle
from app.config import settings
from app.database import SessionLocal
from app import migrate
from app.routers import auth, hosts, logs, rules, groups, aliases, port_forward, search as search_router, users as users_router
from app.routers import auth, hosts, logs, rules, groups, aliases, port_forward, search as search_router, users as users_router, tokens as tokens_router, metrics as metrics_router, mcp as mcp_router, compliance as compliance_router
from app.routers import one_to_one
from app.services.one_to_one_deploy_service import sweep_expired_one_to_one_deployments
from app.routers import outbound_nat
@@ -49,6 +50,7 @@ from app.routers import resource_samples as resource_samples_router
from app.services import resource_monitor_service
from app.routers import scheduled_maintenance as scheduled_maintenance_router
from app.services import scheduled_maintenance_service
from app.services import compliance_service
from app.services import carp_event_service, carp_maintenance_window_service
# Without this, Python's root logger defaults to WARNING with no handler,
@@ -254,6 +256,24 @@ async def _scheduled_maintenance_poll_loop():
await asyncio.sleep(_RUNNING_MAINTENANCE_POLL_SECONDS if in_progress else _SCHEDULED_MAINTENANCE_POLL_SECONDS)
_COMPLIANCE_FIRST_RUN_SECONDS = 120
_COMPLIANCE_INTERVAL_SECONDS = 6 * 3600
async def _compliance_poll_loop():
"""Sixth background loop: refreshes every enabled host's security-posture result.
Each check is one read-only config download per firewall, so every six hours
is plenty; the first run waits a little so a restart is not a thundering herd."""
await asyncio.sleep(_COMPLIANCE_FIRST_RUN_SECONDS)
while True:
try:
async with SessionLocal() as db:
await compliance_service.check_all(db)
except Exception:
logger.exception("Sicherheitslage-Prüfung fehlgeschlagen")
await asyncio.sleep(_COMPLIANCE_INTERVAL_SECONDS)
@asynccontextmanager
async def lifespan(app: FastAPI):
await migrate.run()
@@ -272,6 +292,7 @@ async def lifespan(app: FastAPI):
log_ingest_task = asyncio.create_task(_log_ingest_poll_loop())
resource_task = asyncio.create_task(_resource_poll_loop())
scheduled_maintenance_task = asyncio.create_task(_scheduled_maintenance_poll_loop())
compliance_task = asyncio.create_task(_compliance_poll_loop())
try:
yield
finally:
@@ -280,14 +301,44 @@ async def lifespan(app: FastAPI):
log_ingest_task.cancel()
resource_task.cancel()
scheduled_maintenance_task.cancel()
for task in (sweep_task, notification_task, log_ingest_task, resource_task, scheduled_maintenance_task):
compliance_task.cancel()
for task in (sweep_task, notification_task, log_ingest_task, resource_task, scheduled_maintenance_task, compliance_task):
try:
await task
except asyncio.CancelledError:
pass
app = FastAPI(title="OPNexus", version=APP_VERSION, lifespan=lifespan)
API_DESCRIPTION = (
"REST API of OPNexus, the central management for OPNsense firewall fleets.\n\n"
"**Authentication:** send a personal API token as `Authorization: Bearer opx_...` "
"(create one under *System > API tokens*). Scopes: `read` (GET only), `deploy` "
"(+ roll out rules, NAT, VPN, maintenance), `admin` (+ users, settings, hosts). "
"A token never exceeds its owner's role. Token and account management is session-only.\n\n"
"Errors are `{\"detail\": {\"code\": \"...\", \"params\": {...}}}` with a stable `code`."
)
# The interactive Swagger/ReDoc pages load scripts from a CDN, which the proxy's
# Content-Security-Policy forbids; the schema itself is served (behind auth) for
# any client that renders it (Postman, openapi-generator, an AI assistant).
app = FastAPI(
title="OPNexus", version=APP_VERSION, lifespan=lifespan, description=API_DESCRIPTION,
docs_url=None, redoc_url=None, openapi_url="/api/openapi.json",
)
def _custom_openapi():
if app.openapi_schema is None:
schema = get_openapi(title=app.title, version=app.version, description=app.description, routes=app.routes)
schema.setdefault("components", {}).setdefault("securitySchemes", {})["bearerAuth"] = {
"type": "http", "scheme": "bearer", "description": "Personal API token (opx_...)",
}
schema["security"] = [{"bearerAuth": []}]
app.openapi_schema = schema
return app.openapi_schema
app.openapi = _custom_openapi
@app.exception_handler(RequestValidationError)
async def _safe_validation_error(request: Request, exc: RequestValidationError):
@@ -347,6 +398,10 @@ app.include_router(hosts.router, prefix="/api")
app.include_router(rules.router, prefix="/api")
app.include_router(search_router.router, prefix="/api")
app.include_router(users_router.router, prefix="/api")
app.include_router(tokens_router.router, prefix="/api")
app.include_router(metrics_router.router, prefix="/api")
app.include_router(mcp_router.router, prefix="/api")
app.include_router(compliance_router.router, prefix="/api")
app.include_router(logs.router, prefix="/api")
app.include_router(groups.router, prefix="/api")
app.include_router(aliases.router, prefix="/api")
+38
View File
@@ -0,0 +1,38 @@
"""Personal API tokens for scripts, CI jobs, Prometheus and MCP clients.
Only the SHA-256 of the token is stored (the token is 256 random bits, so a
fast hash is enough and lets a request be resolved with one indexed lookup);
the raw value is shown exactly once, at creation. A token belongs to a user and
never exceeds that user's role: see services/api_token_service.py for the
scopes and AuthMiddleware for where they are enforced.
"""
import uuid
from datetime import datetime
from typing import Optional
from sqlalchemy import DateTime, ForeignKey, String, func
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
class ApiToken(Base):
__tablename__ = "api_tokens"
id: Mapped[uuid.UUID] = mapped_column(PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id: Mapped[uuid.UUID] = mapped_column(
PGUUID(as_uuid=True), ForeignKey("admin_users.id", ondelete="CASCADE"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String(80), nullable=False)
# "read" | "deploy" | "admin"
scope: Mapped[str] = mapped_column(String(10), nullable=False)
token_hash: Mapped[str] = mapped_column(String(64), nullable=False, unique=True, index=True)
# First characters of the token, so a list can tell tokens apart. Not secret.
prefix: Mapped[str] = mapped_column(String(16), nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
expires_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
last_used_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
last_used_ip: Mapped[Optional[str]] = mapped_column(String(45), nullable=True)
user: Mapped["AdminUser"] = relationship("AdminUser")
+31
View File
@@ -0,0 +1,31 @@
"""Latest security-posture result per host (see services/compliance_service.py).
One row per host, replaced on every check. The findings never contain
configuration values beyond rule descriptions and the protocol names they
report on; secrets from config.xml are never read into them.
"""
import uuid
from datetime import datetime
from typing import Optional
from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.orm import Mapped, mapped_column
from app.database import Base
class ComplianceResult(Base):
__tablename__ = "compliance_results"
id: Mapped[uuid.UUID] = mapped_column(PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
host_id: Mapped[uuid.UUID] = mapped_column(
PGUUID(as_uuid=True), ForeignKey("opnsense_hosts.id", ondelete="CASCADE"), nullable=False, unique=True, index=True
)
checked_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
# None when the check could not run (see error).
score: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
findings: Mapped[list] = mapped_column(JSON, nullable=False, default=list)
# None | "unreachable" | "invalid_config"
error: Mapped[Optional[str]] = mapped_column(String(20), nullable=True)
error_detail: Mapped[Optional[str]] = mapped_column(String(300), nullable=True)
+3 -2
View File
@@ -3,7 +3,7 @@
For a fresh install whose generated first-start password scrolled out of the
log, and for an admin who forgot theirs. Needs shell access to the server,
which is the trust boundary for this tool anyway (same as app.reset_2fa).
All sessions of the account are ended. Normally called through the host
All sessions and API tokens of the account are ended. Normally called through the host
CLI:
./opnexus password [username]
@@ -24,7 +24,7 @@ from sqlalchemy import select
from app.config import settings
from app.database import SessionLocal
from app.models.user import AdminUser
from app.services import auth_service
from app.services import api_token_service, auth_service
async def main(username: str | None) -> int:
@@ -48,6 +48,7 @@ async def main(username: str | None) -> int:
password = secrets.token_urlsafe(18)
user.password_hash = await auth_service.hash_password_async(password)
await auth_service.revoke_sessions(db, user.id, commit=False)
await api_token_service.revoke_all(db, user.id, commit=False)
await db.commit()
name = user.username
has_totp = user.totp_enabled
+17 -1
View File
@@ -9,7 +9,7 @@ from app.errors import ErrorCode, api_error
from app.config import settings
from app.database import get_db
from app.models.user import AdminUser
from app.services import auth_service, totp_service
from app.services import api_token_service, auth_service, totp_service
logger = logging.getLogger(__name__)
@@ -63,6 +63,14 @@ def _account_success(user: AdminUser, action: str) -> None:
async def resolve_user(request: Request, db: AsyncSession) -> AdminUser:
# AuthMiddleware already resolved the caller (session cookie or API token), but
# in its own, closed DB session: re-load into this request's session so that
# changes made to the user (2FA, password...) are actually persisted.
resolved = getattr(request.state, "user", None)
if resolved is not None:
user = await db.get(AdminUser, resolved.id)
if user is not None:
return user
token = request.cookies.get(COOKIE_NAME, "")
user = await auth_service.resolve_session(db, token)
if user is None:
@@ -262,6 +270,14 @@ async def logout(request: Request, response: Response, db: AsyncSession = Depend
@router.get("/me")
async def me(request: Request, db: AsyncSession = Depends(get_db)):
raw = api_token_service.bearer_from_header(request.headers.get("authorization", ""))
if raw is not None:
resolved = await api_token_service.resolve(db, raw, client_ip(request))
if resolved is None:
return {"authenticated": False}
user, api_token = resolved
return {"authenticated": True, "username": user.username, "role": user.role,
"via": "token", "scope": api_token.scope, "token_name": api_token.name}
token = request.cookies.get(COOKIE_NAME, "")
user = await auth_service.resolve_session(db, token)
if user is None:
+67
View File
@@ -0,0 +1,67 @@
"""Security posture ("Sicherheitslage"): stored results and on-demand checks.
Reads (GET) only return what the last check stored, so they are cheap and safe
for dashboards, Prometheus and MCP. A check (POST) downloads the host's config
read-only, evaluates it and stores the result; see services/compliance_service.py.
"""
import uuid
from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.errors import ErrorCode, api_error
from app.models.compliance_result import ComplianceResult
from app.models.host import OpnsenseHost
from app.services import compliance_service
router = APIRouter(tags=["compliance"])
def _read(host: OpnsenseHost, row: ComplianceResult | None, *, detail: bool) -> dict:
result = {
"host_id": host.id, "host_name": host.name, "cluster_name": host.cluster_name, "enabled": host.enabled,
"checked_at": row.checked_at if row else None,
"score": row.score if row else None,
"grade": compliance_service.grade(row.score) if row and row.score is not None else None,
"error": row.error if row else None,
"counts": compliance_service.counts(row.findings) if row else None,
}
if detail:
result["findings"] = row.findings if row else []
return result
@router.get("/compliance")
async def fleet_compliance(db: AsyncSession = Depends(get_db)):
hosts = (await db.scalars(select(OpnsenseHost).order_by(OpnsenseHost.name))).all()
rows = {r.host_id: r for r in (await db.scalars(select(ComplianceResult))).all()}
return [_read(host, rows.get(host.id), detail=False) for host in hosts]
@router.post("/compliance/check")
async def check_fleet(db: AsyncSession = Depends(get_db)):
await compliance_service.check_all(db)
return await fleet_compliance(db)
async def _host(db: AsyncSession, host_id: uuid.UUID) -> OpnsenseHost:
host = await db.get(OpnsenseHost, host_id)
if host is None:
raise api_error(404, ErrorCode.HOST_NICHT_GEFUNDEN)
return host
@router.get("/hosts/{host_id}/compliance")
async def host_compliance(host_id: uuid.UUID, db: AsyncSession = Depends(get_db)):
host = await _host(db, host_id)
row = await db.scalar(select(ComplianceResult).where(ComplianceResult.host_id == host_id))
return _read(host, row, detail=True)
@router.post("/hosts/{host_id}/compliance/check")
async def check_host(host_id: uuid.UUID, db: AsyncSession = Depends(get_db)):
host = await _host(db, host_id)
row = await compliance_service.check_host(db, host)
return _read(host, row, detail=True)
+18 -12
View File
@@ -184,18 +184,8 @@ async def _carp_auto_revert_actions(db: AsyncSession) -> list[PendingAction]:
return actions
@router.get("", response_model=DashboardSummary)
async def summary(db: AsyncSession = Depends(get_db)):
hosts = (await db.scalars(select(OpnsenseHost).order_by(OpnsenseHost.name))).all()
statuses = await get_statuses(hosts)
host_summaries = [
DashboardHost(
id=host.id, name=host.name, status=statuses.get(host.id, {}).get("status", "offline"),
carp_status=statuses.get(host.id, {}).get("carp_status"), cluster_name=host.cluster_name,
)
for host in hosts
]
async def collect_pending_actions(db: AsyncSession) -> list[PendingAction]:
"""Everything that needs a human, newest first (shared with the metrics endpoint)."""
pending_actions: list[PendingAction] = []
pending_actions += await _rule_actions(db)
pending_actions += await _nat_actions(db, PortForwardDeployment, PortForwardRule, "port_forward_id", "port_forward", "port-forward")
@@ -209,6 +199,22 @@ async def summary(db: AsyncSession = Depends(get_db)):
pending_actions += await _stuck_host_operations(db)
pending_actions += await _carp_auto_revert_actions(db)
pending_actions.sort(key=lambda item: item.created_at, reverse=True)
return pending_actions
@router.get("", response_model=DashboardSummary)
async def summary(db: AsyncSession = Depends(get_db)):
hosts = (await db.scalars(select(OpnsenseHost).order_by(OpnsenseHost.name))).all()
statuses = await get_statuses(hosts)
host_summaries = [
DashboardHost(
id=host.id, name=host.name, status=statuses.get(host.id, {}).get("status", "offline"),
carp_status=statuses.get(host.id, {}).get("carp_status"), cluster_name=host.cluster_name,
)
for host in hosts
]
pending_actions = await collect_pending_actions(db)
activity_rows = (
await db.scalars(select(AuditEvent).order_by(AuditEvent.created_at.desc(), AuditEvent.id.desc()).limit(15))
+202
View File
@@ -0,0 +1,202 @@
"""Read-only MCP server (Model Context Protocol, streamable HTTP transport).
Lets an AI assistant (Claude Code, Claude Desktop, ...) look at the fleet:
claude mcp add --transport http opnexus https://<opnexus>/api/mcp \
--header "Authorization: Bearer opx_..."
Stateless JSON-RPC over POST (the spec allows plain ``application/json``
replies; there is no server-initiated stream, GET answers 405). Every tool is a
GET against OPNexus' own API, sent through the app itself with the caller's own
credentials, so exactly the caller's permissions apply: a ``read`` token sees
what a ``read`` token sees, secret-bearing endpoints stay closed. Nothing here
can change state.
"""
import json
from urllib.parse import urlencode
import httpx
from fastapi import APIRouter, Request, Response
from fastapi.responses import JSONResponse
from app.version import APP_VERSION
router = APIRouter(tags=["mcp"])
SUPPORTED_VERSIONS = ("2025-06-18", "2025-03-26", "2024-11-05")
MAX_RESULT_CHARS = 60_000
# Not reachable through the generic reader (credentials management, the MCP endpoint itself).
_BLOCKED_PREFIXES = ("/api/tokens", "/api/auth", "/api/mcp")
TOOLS = [
{
"name": "fleet_overview",
"description": "Fleet status at a glance: every firewall with online state and CARP role, items that need "
"attention (failed or unconfirmed deployments, stuck maintenance) and the latest activity.",
"inputSchema": {"type": "object", "properties": {}},
"path": "/api/dashboard",
},
{
"name": "list_hosts",
"description": "All managed firewalls with name, URL, cluster and whether monitoring is enabled.",
"inputSchema": {"type": "object", "properties": {}},
"path": "/api/hosts",
},
{
"name": "host_resources",
"description": "Recent CPU load, memory and disk usage samples of one firewall (a sample per minute).",
"inputSchema": {
"type": "object", "required": ["host_id"],
"properties": {
"host_id": {"type": "string", "description": "Firewall id from list_hosts"},
"since_hours": {"type": "integer", "minimum": 1, "maximum": 720, "default": 1},
},
},
"path": "/api/hosts/{host_id}/resource-samples",
"query": ("since_hours",),
# The raw per-sample payload (every interface counter) would drown the useful numbers.
"drop": ("raw",),
},
{
"name": "carp_events",
"description": "History of CARP role changes (MASTER/BACKUP) across the clusters.",
"inputSchema": {"type": "object", "properties": {}},
"path": "/api/cluster/carp-events",
},
{
"name": "recent_audit",
"description": "The audit trail: who changed or deployed what, newest first.",
"inputSchema": {
"type": "object",
"properties": {
"limit": {"type": "integer", "minimum": 1, "maximum": 200, "default": 25},
"kind": {"type": "string", "description": "e.g. rule, alias, host, user"},
"status": {"type": "string", "description": "e.g. accepted, rejected, denied"},
"q": {"type": "string", "description": "free-text filter"},
},
},
"path": "/api/audit",
"query": ("limit", "kind", "status", "q"),
},
{
"name": "search",
"description": "Search rules, aliases, NAT, VPN, certificates, groups and hosts by text.",
"inputSchema": {
"type": "object", "required": ["q"],
"properties": {"q": {"type": "string", "description": "at least 2 characters"}},
},
"path": "/api/search",
"query": ("q",),
},
{
"name": "api_get",
"description": "Read any other OPNexus API endpoint (GET only), e.g. /api/rules, /api/aliases, /api/logs. "
"The schema of all endpoints is at /api/openapi.json.",
"inputSchema": {
"type": "object", "required": ["path"],
"properties": {
"path": {"type": "string", "description": "Starts with /api/"},
"query": {"type": "object", "description": "Query parameters", "additionalProperties": {"type": "string"}},
},
},
},
]
def _tool_list() -> list[dict]:
return [{k: v for k, v in tool.items() if k in ("name", "description", "inputSchema")} for tool in TOOLS]
def _text(text: str, is_error: bool = False) -> dict:
if len(text) > MAX_RESULT_CHARS:
text = text[:MAX_RESULT_CHARS] + f"\n… truncated ({len(text)} characters in total); narrow the request."
return {"content": [{"type": "text", "text": text}], "isError": is_error}
async def _get(request: Request, path: str, query: dict, drop: tuple = ()) -> dict:
if not path.startswith("/api/") or ".." in path or "?" in path or "#" in path or path.startswith(_BLOCKED_PREFIXES):
return _text("This path cannot be read through MCP.", True)
headers = {k: v for k in ("authorization", "cookie") if (v := request.headers.get(k))}
url = path + (("?" + urlencode(query)) if query else "")
transport = httpx.ASGITransport(app=request.app)
async with httpx.AsyncClient(transport=transport, base_url="http://opnexus", headers=headers, timeout=30) as client:
response = await client.get(url)
body = response.text
try:
data = response.json()
if drop and isinstance(data, list):
data = [{k: v for k, v in item.items() if k not in drop} if isinstance(item, dict) else item for item in data]
body = json.dumps(data, ensure_ascii=False, indent=1)
except ValueError:
pass
return _text(body, response.status_code >= 400)
async def _call_tool(request: Request, name: str, arguments: dict) -> dict:
tool = next((t for t in TOOLS if t["name"] == name), None)
if tool is None:
return _text(f"Unknown tool: {name}", True)
if name == "api_get":
query = arguments.get("query") or {}
if not isinstance(arguments.get("path"), str) or not isinstance(query, dict):
return _text("path (string) is required.", True)
return await _get(request, arguments["path"], {str(k): str(v) for k, v in query.items()})
path = tool["path"]
if "{host_id}" in path:
host_id = str(arguments.get("host_id", ""))
if not host_id.replace("-", "").isalnum():
return _text("host_id is required (see list_hosts).", True)
path = path.format(host_id=host_id)
query = {key: arguments[key] for key in tool.get("query", ()) if arguments.get(key) not in (None, "")}
return await _get(request, path, query, tool.get("drop", ()))
def _error(request_id, code: int, message: str) -> dict:
return {"jsonrpc": "2.0", "id": request_id, "error": {"code": code, "message": message}}
async def _handle(request: Request, message: dict) -> dict | None:
"""The JSON-RPC response for one message, None for a notification."""
if not isinstance(message, dict) or message.get("jsonrpc") != "2.0" or "method" not in message:
return _error(message.get("id") if isinstance(message, dict) else None, -32600, "Invalid request")
method, request_id = message["method"], message.get("id")
params = message.get("params") or {}
if "id" not in message: # notification (e.g. notifications/initialized): nothing to answer
return None
if method == "initialize":
requested = params.get("protocolVersion")
return {"jsonrpc": "2.0", "id": request_id, "result": {
"protocolVersion": requested if requested in SUPPORTED_VERSIONS else SUPPORTED_VERSIONS[0],
"capabilities": {"tools": {"listChanged": False}},
"serverInfo": {"name": "opnexus", "version": APP_VERSION},
"instructions": "Read-only access to an OPNsense fleet managed by OPNexus.",
}}
if method == "ping":
return {"jsonrpc": "2.0", "id": request_id, "result": {}}
if method == "tools/list":
return {"jsonrpc": "2.0", "id": request_id, "result": {"tools": _tool_list()}}
if method == "tools/call":
arguments = params.get("arguments") or {}
if not isinstance(params.get("name"), str) or not isinstance(arguments, dict):
return _error(request_id, -32602, "name (string) and arguments (object) are required")
return {"jsonrpc": "2.0", "id": request_id, "result": await _call_tool(request, params["name"], arguments)}
return _error(request_id, -32601, f"Method not found: {method}")
@router.post("/mcp")
async def mcp_endpoint(request: Request):
try:
payload = await request.json()
except ValueError:
return JSONResponse(_error(None, -32700, "Parse error"), status_code=400)
if isinstance(payload, list):
replies = [reply for message in payload if (reply := await _handle(request, message)) is not None]
return JSONResponse(replies) if replies else Response(status_code=202)
reply = await _handle(request, payload)
return JSONResponse(reply) if reply is not None else Response(status_code=202)
@router.get("/mcp")
async def mcp_no_stream():
# No server-initiated messages: tell clients not to expect an SSE stream.
return Response(status_code=405, headers={"Allow": "POST"})
+18
View File
@@ -0,0 +1,18 @@
"""GET /api/metrics — Prometheus text exposition (see services/metrics_service.py).
Authenticated like everything else; the intended client is a Prometheus job
with ``authorization: {credentials: <read token>}``.
"""
from fastapi import APIRouter, Depends
from fastapi.responses import PlainTextResponse
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.services import metrics_service
router = APIRouter(tags=["metrics"])
@router.get("/metrics", response_class=PlainTextResponse)
async def metrics(db: AsyncSession = Depends(get_db)):
return PlainTextResponse(await metrics_service.render(db), media_type=metrics_service.CONTENT_TYPE)
+67
View File
@@ -0,0 +1,67 @@
"""Manage personal API tokens (see services/api_token_service.py for scopes).
Session-only: an API token cannot reach these endpoints (AuthMiddleware), so a
leaked token can never mint a replacement or list/revoke its siblings. A
read_only user may manage their own tokens, but only with read scope.
"""
import uuid
from fastapi import APIRouter, Depends, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.errors import ErrorCode, api_error
from app.models.api_token import ApiToken
from app.models.user import AdminUser
from app.routers.auth import _account_failure, _account_success, _check_account_budget, current_user
from app.schemas.api_token import ApiTokenCreate, ApiTokenCreated, ApiTokenRead, TokenScope
from app.services import api_token_service, auth_service
router = APIRouter(prefix="/tokens", tags=["tokens"])
def _read(token: ApiToken, owner: str) -> dict:
return {
"id": token.id, "name": token.name, "scope": token.scope, "prefix": token.prefix, "owner": owner,
"created_at": token.created_at, "expires_at": token.expires_at,
"last_used_at": token.last_used_at, "last_used_ip": token.last_used_ip,
}
@router.get("", response_model=list[ApiTokenRead])
async def list_tokens(db: AsyncSession = Depends(get_db), user: AdminUser = Depends(current_user)):
"""Own tokens; an admin sees everybody's (to be able to revoke them)."""
query = select(ApiToken, AdminUser.username).join(AdminUser, ApiToken.user_id == AdminUser.id)
if user.role != "admin":
query = query.where(ApiToken.user_id == user.id)
rows = (await db.execute(query.order_by(ApiToken.created_at.desc()))).all()
return [_read(token, owner) for token, owner in rows]
@router.post("", response_model=ApiTokenCreated, status_code=status.HTTP_201_CREATED)
async def create_token(
payload: ApiTokenCreate, db: AsyncSession = Depends(get_db), user: AdminUser = Depends(current_user)
):
if user.role != "admin" and payload.scope != TokenScope.read:
raise api_error(403, ErrorCode.API_TOKEN_SCOPE_TOO_HIGH)
_check_account_budget(user, "create-token")
if not await auth_service.verify_password_async(user.password_hash, payload.current_password):
_account_failure(user, "create-token")
raise api_error(401, ErrorCode.AKTUELLES_PASSWORT_IST_FALSCH)
_account_success(user, "create-token")
token, raw = await api_token_service.create(
db, user, name=payload.name.strip(), scope=payload.scope.value, expires_in_days=payload.expires_in_days
)
return {**_read(token, user.username), "token": raw}
@router.delete("/{token_id}", status_code=204)
async def revoke_token(
token_id: uuid.UUID, db: AsyncSession = Depends(get_db), user: AdminUser = Depends(current_user)
):
token = await db.get(ApiToken, token_id)
if token is None or (token.user_id != user.id and user.role != "admin"):
raise api_error(404, ErrorCode.API_TOKEN_NOT_FOUND)
await db.delete(token)
await db.commit()
+2 -1
View File
@@ -9,7 +9,7 @@ from app.database import get_db
from app.models.user import AdminUser
from app.routers.auth import current_user
from app.schemas.user import UserCreate, UserPasswordReset, UserRead, UserRoleUpdate
from app.services import auth_service, totp_service
from app.services import api_token_service, auth_service, totp_service
router = APIRouter(prefix="/users", tags=["users"])
@@ -79,6 +79,7 @@ async def reset_password(user_id: uuid.UUID, payload: UserPasswordReset, db: Asy
# A reset is the answer to a suspected compromise: every session of the
# account ends with the old password, atomically with the new hash.
await auth_service.revoke_sessions(db, target.id, commit=False)
await api_token_service.revoke_all(db, target.id, commit=False)
await db.commit()
return target
+40
View File
@@ -0,0 +1,40 @@
import uuid
from datetime import datetime
from enum import Enum
from pydantic import BaseModel, ConfigDict, Field
class TokenScope(str, Enum):
read = "read"
deploy = "deploy"
admin = "admin"
class ApiTokenCreate(BaseModel):
name: str = Field(min_length=1, max_length=80)
# Least privilege by default.
scope: TokenScope = TokenScope.read
# None = never expires. The UI defaults to 90 days.
expires_in_days: int | None = Field(default=None, ge=1, le=3650)
# Creating a long-lived credential needs the account password again.
current_password: str = Field(min_length=1, max_length=1024)
class ApiTokenRead(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
name: str
scope: TokenScope
prefix: str
owner: str
created_at: datetime
expires_at: datetime | None
last_used_at: datetime | None
last_used_ip: str | None
class ApiTokenCreated(ApiTokenRead):
# The only time the raw token is ever returned.
token: str
+126
View File
@@ -0,0 +1,126 @@
"""Personal API tokens: creation, lookup and scope rules.
Scopes (a token can never do more than its owner's role allows; a read_only
user's token is read-only whatever its scope says):
* ``read`` — GET/HEAD/OPTIONS only, except endpoints that return secrets (raw
config backups, notification settings): those need ``admin``. For Prometheus,
dashboards, MCP clients.
* ``deploy`` — additionally every write that rolls out or operates on the fleet
(rules, aliases, NAT, VPN, maintenance, backups...). NOT the writes that
change who can access OPNexus or where it sends data: users, settings,
notifications, host registration/removal, log settings, OPNexus self-update.
* ``admin`` — everything the owner may do, except what is session-only.
Session-only for every scope: managing tokens themselves and the account
endpoints (/api/auth/*), so a leaked token can neither mint more tokens nor
change a password or second factor.
The enforcement lives in AuthMiddleware (one place, like the read_only role).
"""
import hashlib
import re
import secrets
from datetime import datetime, timedelta, timezone
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.errors import ErrorCode, api_error
from app.models.api_token import ApiToken
from app.models.user import AdminUser
TOKEN_PREFIX = "opx_"
SCOPES = ("read", "deploy", "admin")
MAX_TOKENS_PER_USER = 50
# last_used_at is only rewritten when older than this, so a scraper hitting
# /api/metrics every few seconds does not turn into a write per request.
_LAST_USED_GRANULARITY = timedelta(minutes=1)
_SAFE_METHODS = ("GET", "HEAD", "OPTIONS")
_SESSION_ONLY = re.compile(r"^/api/(tokens|auth)(/|$)")
# Reads that expose secrets (config.xml holds password hashes and VPN keys, the
# notification settings the webhook URL/SMTP data): admin scope only.
_ADMIN_SCOPE_READS = re.compile(r"^/api/hosts/[^/]+/backups/[^/]+$|^/api/notifications(/|$)")
# The MCP endpoint is POST (JSON-RPC) but its tools only ever read.
_MCP_PATH = "/api/mcp"
_ADMIN_ONLY_WRITES = re.compile(
r"^/api/(users|settings|system|notifications)(/|$)"
r"|^/api/logs/settings$"
r"|^/api/hosts(/tls-probe|/[^/]+)?$"
)
def _now() -> datetime:
return datetime.now(timezone.utc)
def hash_token(raw: str) -> str:
return hashlib.sha256(raw.encode()).hexdigest()
def denies(scope: str, method: str, path: str) -> bool:
"""True if a token with this scope must not make this request."""
if _SESSION_ONLY.match(path):
return True
if path == _MCP_PATH:
return False
if method in _SAFE_METHODS:
return scope != "admin" and bool(_ADMIN_SCOPE_READS.match(path))
if scope == "read":
return True
if scope == "deploy" and _ADMIN_ONLY_WRITES.match(path):
return True
return False
def bearer_from_header(header: str) -> str | None:
"""The token of an ``Authorization: Bearer opx_...`` header, else None."""
scheme, _, value = header.partition(" ")
value = value.strip()
if scheme.lower() == "bearer" and value.startswith(TOKEN_PREFIX):
return value
return None
async def create(
db: AsyncSession, user: AdminUser, *, name: str, scope: str, expires_in_days: int | None
) -> tuple[ApiToken, str]:
count = len((await db.scalars(select(ApiToken.id).where(ApiToken.user_id == user.id))).all())
if count >= MAX_TOKENS_PER_USER:
raise api_error(409, ErrorCode.API_TOKEN_LIMIT_REACHED, limit=MAX_TOKENS_PER_USER)
raw = TOKEN_PREFIX + secrets.token_urlsafe(32)
token = ApiToken(
user_id=user.id,
name=name,
scope=scope,
token_hash=hash_token(raw),
prefix=raw[: len(TOKEN_PREFIX) + 6],
expires_at=_now() + timedelta(days=expires_in_days) if expires_in_days else None,
)
db.add(token)
await db.commit()
await db.refresh(token)
return token, raw
async def resolve(db: AsyncSession, raw: str, ip: str | None) -> tuple[AdminUser, ApiToken] | None:
"""The owner and token for a raw bearer token, or None if unknown/expired."""
token = await db.scalar(select(ApiToken).where(ApiToken.token_hash == hash_token(raw)))
if token is None or (token.expires_at is not None and token.expires_at <= _now()):
return None
user = await db.get(AdminUser, token.user_id)
if user is None:
return None
if token.last_used_at is None or _now() - token.last_used_at > _LAST_USED_GRANULARITY:
token.last_used_at = _now()
token.last_used_ip = ip
await db.commit()
return user, token
async def revoke_all(db: AsyncSession, user_id, *, commit: bool = True) -> None:
"""Delete every token of an account (password reset: answer to a suspected compromise)."""
await db.execute(delete(ApiToken).where(ApiToken.user_id == user_id))
if commit:
await db.commit()
+1 -1
View File
@@ -1,7 +1,7 @@
"""Authentication core: password hashing, server-side sessions, brute-force
protection and the initial admin bootstrap.
Two roles ("admin", "read_only"), no orgs, no API keys — see
Two roles ("admin", "read_only"), no orgs; API tokens live in api_token_service — see
AdminUser/AuthMiddleware for how a role is enforced. Originally mirrored
InfraPanel's own app/services/auth_service.py (single-admin-only); the role
column and its enforcement are OPNexus-specific additions on top of that.
+276
View File
@@ -0,0 +1,276 @@
"""Security posture of a firewall: hardening checks and rule hygiene.
The checks read the firewall's own config.xml (one read-only download, parsed
in memory with defusedxml, never stored) plus two facts OPNexus already knows:
whether firmware updates are pending and how old the newest local backup is.
The XML layout was verified against real OPNsense 26.7 configs: rules live in
``OPNsense/Firewall/Filter/rules/rule`` (MVC) or, on installs not yet migrated,
in ``filter/rule`` (legacy). Both are normalised to the same shape below.
``evaluate`` is pure (bytes in, findings out). A finding is
``{id, severity, passed, params}``; ``passed`` is None when a check could not be
decided. The score starts at 100 and loses a penalty per failed check, so one
critical finding (an any-any pass rule on the WAN) already turns the host red.
"""
import re
from datetime import datetime, timezone
from xml.etree.ElementTree import Element
import defusedxml.ElementTree as safe_ET
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.compliance_result import ComplianceResult
from app.models.host import OpnsenseHost
from app.services import backup_store
from app.services import opnsense_client as client
from app.services.opnsense_client import OpnsenseAPIError
MAX_CONFIG_BYTES = 32 * 1024 * 1024
BACKUP_MAX_AGE_DAYS = 7
PENALTY = {"critical": 30, "high": 15, "medium": 8, "low": 3, "info": 0}
SEVERITY_ORDER = ("critical", "high", "medium", "low", "info")
MAX_EXAMPLES = 5
def grade(score: int) -> str:
return "good" if score >= 90 else "fair" if score >= 70 else "poor"
def _text(el: Element | None, path: str) -> str | None:
node = el.find(path) if el is not None else None
if node is None or node.text is None or not node.text.strip():
return None
return node.text.strip()
# ── Rule normalisation ───────────────────────────────────────────────────
def _endpoint(node: Element | None) -> tuple[str, bool, str | None]:
"""(network, negated, port) of a legacy <source>/<destination> element."""
if node is None:
return "any", False, None
negated = node.find("not") is not None
port = _text(node, "port")
if node.find("any") is not None:
return "any", negated, port
for tag in ("network", "address"):
value = _text(node, tag)
if value:
return value, negated, port
return "any", negated, port
def _mvc_rule(rule: Element) -> dict:
return {
"enabled": _text(rule, "enabled") != "0",
"action": (_text(rule, "action") or "pass").lower(),
"interfaces": [i for i in (_text(rule, "interface") or "").split(",") if i],
"direction": _text(rule, "direction") or "in",
"ipprotocol": _text(rule, "ipprotocol") or "inet",
"protocol": (_text(rule, "protocol") or "any").lower(),
"src": _text(rule, "source_net") or "any", "src_not": _text(rule, "source_not") == "1",
"sport": _text(rule, "source_port"),
"dst": _text(rule, "destination_net") or "any", "dst_not": _text(rule, "destination_not") == "1",
"dport": _text(rule, "destination_port"),
"floating": False,
"quick": _text(rule, "quick") != "0",
"descr": _text(rule, "description") or "",
"order": int(_text(rule, "sequence") or 0),
}
def _legacy_rule(rule: Element, index: int) -> dict:
src, src_not, sport = _endpoint(rule.find("source"))
dst, dst_not, dport = _endpoint(rule.find("destination"))
return {
"enabled": rule.find("disabled") is None,
"action": (_text(rule, "type") or "pass").lower(),
"interfaces": [i for i in (_text(rule, "interface") or "").split(",") if i],
"direction": _text(rule, "direction") or "in",
"ipprotocol": _text(rule, "ipprotocol") or "inet",
"protocol": (_text(rule, "protocol") or "any").lower(),
"src": src, "src_not": src_not, "sport": sport, "dst": dst, "dst_not": dst_not, "dport": dport,
"floating": _text(rule, "floating") == "yes",
"quick": _text(rule, "quick") != "0",
"descr": _text(rule, "descr") or "",
"order": index,
}
def extract_rules(root: Element) -> list[list[dict]]:
"""Rule lists in evaluation order, one per source (MVC, legacy)."""
mvc = sorted((_mvc_rule(r) for r in root.findall("OPNsense/Firewall/Filter/rules/rule")), key=lambda r: r["order"])
legacy = [_legacy_rule(r, i) for i, r in enumerate(root.findall("filter/rule"))]
return [group for group in (mvc, legacy) if group]
def wan_interfaces(root: Element) -> set[str]:
"""Interfaces facing the outside: the one named wan*, or any with an upstream gateway."""
result = set()
for node in root.findall("interfaces/*"):
if node.tag.lower().startswith("wan") or _text(node, "gateway"):
result.add(node.tag)
return result
def _is_any_any(rule: dict) -> bool:
return (
rule["enabled"] and rule["action"] == "pass" and rule["src"] == "any" and not rule["src_not"]
and rule["dst"] == "any" and not rule["dst_not"] and rule["protocol"] == "any"
and not rule["sport"] and not rule["dport"]
)
def _covers(a: dict, b: dict) -> bool:
"""Every packet matched by rule b is already matched by the earlier rule a.
Only exact-or-any comparisons, so it never claims coverage it cannot prove."""
if not (a["enabled"] and b["enabled"]) or a["floating"] or b["floating"] or not a["quick"]:
return False
if not a["interfaces"] or not set(b["interfaces"]) <= set(a["interfaces"]):
return False
if a["direction"] != b["direction"] or a["ipprotocol"] != b["ipprotocol"]:
return False
if a["protocol"] != "any" and a["protocol"] != b["protocol"]:
return False
for side in ("src", "dst"):
if a[side + "_not"] or b[side + "_not"]:
if (a[side], a[side + "_not"]) != (b[side], b[side + "_not"]):
return False
elif a[side] != "any" and a[side] != b[side]:
return False
for port in ("sport", "dport"):
if a[port] and a[port] != b[port]:
return False
return True
def _label(rule: dict) -> str:
return rule["descr"] or f"{rule['action']} {rule['src']} → {rule['dst']}"
def analyse_rules(groups: list[list[dict]]) -> dict:
shadowed, duplicates = [], []
for rules in groups:
for j, later in enumerate(rules):
for earlier in rules[:j]:
if _covers(earlier, later):
(duplicates if earlier["action"] == later["action"] and _covers(later, earlier) else shadowed).append(_label(later))
break
return {"shadowed": shadowed, "duplicates": duplicates}
# ── Checks ───────────────────────────────────────────────────────────────
def _finding(check_id: str, severity: str, passed: bool | None, **params) -> dict:
return {"id": check_id, "severity": severity, "passed": passed, "params": params}
def _listing(items: list[str]) -> dict:
return {"count": len(items), "examples": items[:MAX_EXAMPLES]}
def evaluate(xml_bytes: bytes, *, firmware_status: str | None, newest_backup_age_s: float | None) -> list[dict]:
if len(xml_bytes) > MAX_CONFIG_BYTES or not xml_bytes.strip():
raise ValueError("config.xml missing or too large")
root = safe_ET.fromstring(xml_bytes)
if root.tag != "opnsense":
raise ValueError("not an OPNsense config.xml")
findings: list[dict] = []
# Management access
protocol = _text(root, "system/webgui/protocol")
findings.append(_finding("webgui_https", "high", protocol == "https", protocol=protocol or "http"))
ssh = root.find("system/ssh")
ssh_enabled = ssh is not None and _text(ssh, "enabled") in ("enabled", "1")
root_login = ssh_enabled and _text(ssh, "permitrootlogin") == "1"
password_login = ssh_enabled and _text(ssh, "passwordauth") == "1"
if root_login and password_login:
findings.append(_finding("ssh_hardened", "high", False, reason="root_password"))
elif password_login:
findings.append(_finding("ssh_hardened", "medium", False, reason="password"))
else:
findings.append(_finding("ssh_hardened", "high", True))
community = _text(root, "snmpd/rocommunity")
if root.find("snmpd") is not None:
findings.append(_finding("snmp_community", "high", (community or "").lower() not in ("", "public")))
# Rules
wans = wan_interfaces(root)
groups = extract_rules(root)
all_rules = [rule for group in groups for rule in group]
wan_open = [_label(r) for r in all_rules if _is_any_any(r) and set(r["interfaces"]) & wans]
findings.append(_finding("wan_any_any", "critical", not wan_open, **_listing(wan_open)))
inner_open = [_label(r) for r in all_rules if _is_any_any(r) and r["interfaces"] and not set(r["interfaces"]) & wans]
findings.append(_finding("internal_any_any", "medium", not inner_open, **_listing(inner_open)))
analysis = analyse_rules(groups)
findings.append(_finding("rules_shadowed", "medium", not analysis["shadowed"], **_listing(analysis["shadowed"])))
findings.append(_finding("rules_duplicate", "low", not analysis["duplicates"], **_listing(analysis["duplicates"])))
undocumented = [f"{r['action']} {r['src']} → {r['dst']}" for r in all_rules if r["enabled"] and not r["descr"]]
findings.append(_finding("rules_documented", "low", not undocumented, **_listing(undocumented)))
# Operations
findings.append(_finding("ntp_configured", "low", bool(_text(root, "system/timeservers"))))
syslog_targets = [d for d in root.findall("OPNsense/Syslog/destinations/destination") if _text(d, "enabled") != "0"]
findings.append(_finding("remote_logging", "low", bool(syslog_targets) or bool(_text(root, "syslog/remoteserver"))))
findings.append(_finding("ids_enabled", "info", _text(root, "OPNsense/IDS/general/enabled") == "1"))
# From what OPNexus already knows
findings.append(_finding("firmware_current", "medium", None if firmware_status is None else firmware_status != "update"))
if newest_backup_age_s is None:
findings.append(_finding("backup_recent", "medium", False, age_days=None))
else:
age_days = int(newest_backup_age_s // 86400)
findings.append(_finding("backup_recent", "medium", age_days < BACKUP_MAX_AGE_DAYS, age_days=age_days))
return findings
def score(findings: list[dict]) -> int:
return max(0, 100 - sum(PENALTY[f["severity"]] for f in findings if f["passed"] is False))
def counts(findings: list[dict]) -> dict:
result = {severity: 0 for severity in SEVERITY_ORDER}
for finding in findings:
if finding["passed"] is False:
result[finding["severity"]] += 1
return result
# ── Running it for a host ────────────────────────────────────────────────
async def check_host(db: AsyncSession, host: OpnsenseHost) -> ComplianceResult:
"""Download the config, evaluate, store (one row per host, replaced each time)."""
row = await db.scalar(select(ComplianceResult).where(ComplianceResult.host_id == host.id))
if row is None:
row = ComplianceResult(host_id=host.id)
db.add(row)
row.checked_at = datetime.now(timezone.utc)
try:
xml = await client.download_backup(host)
try:
firmware_status = (await client.get_firmware_status(host)).get("status")
except OpnsenseAPIError:
firmware_status = None
snapshots = backup_store.list_snapshots(host.id)
age = (datetime.now(timezone.utc) - snapshots[0]["created_at"]).total_seconds() if snapshots else None
findings = evaluate(xml, firmware_status=firmware_status, newest_backup_age_s=age)
row.findings, row.score, row.error = findings, score(findings), None
except OpnsenseAPIError as exc:
row.findings, row.score, row.error = [], None, "unreachable"
row.error_detail = str(exc)[:300]
except ValueError as exc:
row.findings, row.score, row.error = [], None, "invalid_config"
row.error_detail = str(exc)[:300]
await db.commit()
return row
async def check_all(db: AsyncSession) -> int:
hosts = (await db.scalars(select(OpnsenseHost).where(OpnsenseHost.enabled.is_(True)).order_by(OpnsenseHost.name))).all()
for host in hosts:
await check_host(db, host)
return len(hosts)
+155
View File
@@ -0,0 +1,155 @@
"""Prometheus exposition of the fleet state (GET /api/metrics).
Deliberately database-only, like the dashboard summary: a scrape must never
fan out live calls to every firewall (an offline host would stall it past the
scraper's timeout and a 15 s scrape interval would hammer the fleet). The
inputs are what OPNexus' own pollers already store:
* host reachability and resources — the 60 s resource sampler (a host that
produced no sample for ``STALE_AFTER`` is reported down),
* CARP role — the latest observed transition (carp_events),
* open problems — the notification engine's problem table,
* pending actions — the same list the dashboard shows.
"""
from datetime import datetime, timedelta, timezone
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.carp_event import CarpEvent
from app.models.compliance_result import ComplianceResult
from app.models.host import OpnsenseHost
from app.models.notification import NotificationProblem
from app.models.resource_sample import HostResourceSample
from app.models.scheduled_maintenance import ScheduledMaintenance
from app.routers.dashboard import collect_pending_actions
from app.services import compliance_service
from app.version import APP_VERSION
CONTENT_TYPE = "text/plain; version=0.0.4; charset=utf-8"
# The sampler runs every 60 s; three missed rounds mean the host is not reachable.
STALE_AFTER = timedelta(seconds=180)
def _escape(value: str) -> str:
return value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
class _Exposition:
def __init__(self) -> None:
self.lines: list[str] = []
def family(self, name: str, help_text: str, kind: str = "gauge") -> None:
self.lines.append(f"# HELP {name} {help_text}")
self.lines.append(f"# TYPE {name} {kind}")
def sample(self, name: str, value: float, **labels: str) -> None:
rendered = ",".join(f'{key}="{_escape(str(val))}"' for key, val in labels.items() if val is not None)
number = repr(float(value)) if not float(value).is_integer() else str(int(value))
self.lines.append(f"{name}{{{rendered}}} {number}" if rendered else f"{name} {number}")
def text(self) -> str:
return "\n".join(self.lines) + "\n"
async def render(db: AsyncSession) -> str:
now = datetime.now(timezone.utc)
out = _Exposition()
out.family("opnexus_info", "OPNexus version (always 1).")
out.sample("opnexus_info", 1, version=APP_VERSION)
hosts = (await db.scalars(select(OpnsenseHost).order_by(OpnsenseHost.name))).all()
latest_sample = (
select(HostResourceSample.host_id, func.max(HostResourceSample.sampled_at).label("at"))
.group_by(HostResourceSample.host_id)
.subquery()
)
samples = {
sample.host_id: sample
for sample in (
await db.scalars(
select(HostResourceSample).join(
latest_sample,
(HostResourceSample.host_id == latest_sample.c.host_id)
& (HostResourceSample.sampled_at == latest_sample.c.at),
)
)
).all()
}
latest_event = (
select(CarpEvent.host_id, func.max(CarpEvent.detected_at).label("at")).group_by(CarpEvent.host_id).subquery()
)
carp_state = {
event.host_id: event.to_status
for event in (
await db.scalars(
select(CarpEvent).join(
latest_event, (CarpEvent.host_id == latest_event.c.host_id) & (CarpEvent.detected_at == latest_event.c.at)
)
)
).all()
}
out.family("opnexus_hosts", "Number of managed firewalls.")
out.sample("opnexus_hosts", len(hosts))
out.family("opnexus_host_enabled", "1 if monitoring of the host is enabled.")
out.family("opnexus_host_up", "1 if the host answered the resource sampler within the last 3 minutes.")
out.family("opnexus_host_last_sample_timestamp_seconds", "Unix time of the last successful resource sample.")
out.family("opnexus_host_memory_used_percent", "Memory in use, percent (last sample).")
out.family("opnexus_host_load1", "1-minute load average (last sample).")
out.family("opnexus_host_disk_used_percent", "Fullest disk, percent (last sample).")
out.family("opnexus_host_carp_master", "1 if the host's last observed CARP role is MASTER, 0 if BACKUP; absent if unknown.")
for host in hosts:
labels = {"host": host.name, "cluster": host.cluster_name or ""}
sample = samples.get(host.id)
up = bool(host.enabled and sample is not None and now - sample.sampled_at <= STALE_AFTER)
out.sample("opnexus_host_enabled", 1 if host.enabled else 0, **labels)
out.sample("opnexus_host_up", 1 if up else 0, **labels)
if sample is not None:
out.sample("opnexus_host_last_sample_timestamp_seconds", sample.sampled_at.timestamp(), **labels)
for name, value in (
("opnexus_host_memory_used_percent", sample.mem_used_pct),
("opnexus_host_load1", sample.load_avg_1m),
("opnexus_host_disk_used_percent", sample.disk_used_pct),
):
if value is not None:
out.sample(name, value, **labels)
if carp_state.get(host.id) in ("MASTER", "BACKUP"):
out.sample("opnexus_host_carp_master", 1 if carp_state[host.id] == "MASTER" else 0, **labels)
out.family("opnexus_host_compliance_score", "Security posture score 0-100 of the last check (absent if never checked).")
out.family("opnexus_host_compliance_findings", "Failed security checks of the last check, by severity.")
results = {r.host_id: r for r in (await db.scalars(select(ComplianceResult))).all()}
for host in hosts:
row = results.get(host.id)
if row is None or row.score is None:
continue
labels = {"host": host.name, "cluster": host.cluster_name or ""}
out.sample("opnexus_host_compliance_score", row.score, **labels)
for severity, count in compliance_service.counts(row.findings).items():
out.sample("opnexus_host_compliance_findings", count, severity=severity, **labels)
out.family("opnexus_pending_actions", "Deployments and operations that need attention, by kind and status.")
counts: dict[tuple[str, str], int] = {}
for action in await collect_pending_actions(db):
counts[(action.kind, action.status)] = counts.get((action.kind, action.status), 0) + 1
for (kind, status), count in sorted(counts.items()):
out.sample("opnexus_pending_actions", count, kind=kind, status=status)
out.family("opnexus_problems_open", "Open problems known to the notification engine, by type.")
rows = (
await db.execute(
select(NotificationProblem.problem_type, func.count())
.where(NotificationProblem.status == "open")
.group_by(NotificationProblem.problem_type)
)
).all()
for problem_type, count in sorted(rows):
out.sample("opnexus_problems_open", count, type=problem_type)
out.family("opnexus_scheduled_maintenance_pending", "Scheduled maintenance jobs that have not run yet.")
pending = await db.scalar(select(func.count()).select_from(ScheduledMaintenance).where(ScheduledMaintenance.status == "pending"))
out.sample("opnexus_scheduled_maintenance_pending", pending or 0)
return out.text()
+2 -2
View File
@@ -6,5 +6,5 @@ convention — but both are bumped together whenever a release ships).
Versioning scheme: see CHANGELOG.md ("Versionierung").
"""
APP_VERSION = "1.17.0"
APP_CODENAME = "Imperial"
APP_VERSION = "1.18.0"
APP_CODENAME = "Kestrel"
@@ -0,0 +1,42 @@
"""API-Tokens für Skripte, CI, Prometheus und MCP.
Revision ID: 0008
Revises: 0007
Create Date: 2026-10-19
"""
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
revision = "0008"
down_revision = "0007"
branch_labels = None
depends_on = None
def upgrade() -> None:
# Eine Altinstallation (vor Alembic) wird mit create_all() nach den aktuellen
# Modellen gebaut und hat die Tabelle dann schon.
if sa.inspect(op.get_bind()).has_table("api_tokens"):
return
op.create_table(
"api_tokens",
sa.Column("id", postgresql.UUID(as_uuid=True), nullable=False),
sa.Column("user_id", postgresql.UUID(as_uuid=True), nullable=False),
sa.Column("name", sa.String(length=80), nullable=False),
sa.Column("scope", sa.String(length=10), nullable=False),
sa.Column("token_hash", sa.String(length=64), nullable=False),
sa.Column("prefix", sa.String(length=16), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("last_used_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("last_used_ip", sa.String(length=45), nullable=True),
sa.ForeignKeyConstraint(["user_id"], ["admin_users.id"], ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id"),
)
op.create_index("ix_api_tokens_user_id", "api_tokens", ["user_id"])
op.create_index("ix_api_tokens_token_hash", "api_tokens", ["token_hash"], unique=True)
def downgrade() -> None:
op.execute("DROP TABLE IF EXISTS api_tokens")
@@ -0,0 +1,37 @@
"""Sicherheitslage: letztes Prüfergebnis je Host.
Revision ID: 0009
Revises: 0008
Create Date: 2026-10-19
"""
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
revision = "0009"
down_revision = "0008"
branch_labels = None
depends_on = None
def upgrade() -> None:
# Eine Altinstallation wird mit create_all() nach den aktuellen Modellen gebaut.
if sa.inspect(op.get_bind()).has_table("compliance_results"):
return
op.create_table(
"compliance_results",
sa.Column("id", postgresql.UUID(as_uuid=True), nullable=False),
sa.Column("host_id", postgresql.UUID(as_uuid=True), nullable=False),
sa.Column("checked_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("score", sa.Integer(), nullable=True),
sa.Column("findings", sa.JSON(), nullable=False),
sa.Column("error", sa.String(length=20), nullable=True),
sa.Column("error_detail", sa.String(length=300), nullable=True),
sa.ForeignKeyConstraint(["host_id"], ["opnsense_hosts.id"], ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id"),
)
op.create_index("ix_compliance_results_host_id", "compliance_results", ["host_id"], unique=True)
def downgrade() -> None:
op.execute("DROP TABLE IF EXISTS compliance_results")
+281
View File
@@ -0,0 +1,281 @@
"""API tokens: creation, scopes, session-only boundaries, revocation, audit."""
from datetime import datetime, timedelta, timezone
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import select, update
from app.config import settings
from app.database import SessionLocal
from app.main import app
from app.models.api_token import ApiToken
from app.services import api_token_service
PASSWORD = settings.admin_password
async def _make(client, scope="read", name="ci", **extra):
resp = await client.post("/api/tokens", json={"name": name, "scope": scope, "current_password": PASSWORD, **extra})
assert resp.status_code == 201, resp.text
return resp.json()
def _bearer(token: str) -> AsyncClient:
return AsyncClient(
transport=ASGITransport(app=app), base_url="http://test", headers={"Authorization": f"Bearer {token}"}
)
async def _login(username: str, password: str) -> AsyncClient:
ac = AsyncClient(transport=ASGITransport(app=app), base_url="http://test")
resp = await ac.post("/api/auth/login", json={"username": username, "password": password})
assert resp.status_code == 200, resp.text
return ac
@pytest.mark.parametrize(
"scope,method,path,denied",
[
("read", "GET", "/api/hosts", False),
("read", "POST", "/api/rules", True),
# Secret-bearing reads need the admin scope; MCP (read-only tools) is open to every scope.
("read", "GET", "/api/hosts/abc/backups/config-1.xml", True),
("deploy", "GET", "/api/hosts/abc/backups/config-1.xml", True),
("admin", "GET", "/api/hosts/abc/backups/config-1.xml", False),
("read", "GET", "/api/hosts/abc/backups", False),
("read", "GET", "/api/notifications/settings", True),
("admin", "GET", "/api/notifications/settings", False),
("read", "POST", "/api/mcp", False),
("deploy", "POST", "/api/rules", False),
("deploy", "DELETE", "/api/aliases/x", False),
("deploy", "POST", "/api/hosts", True),
("deploy", "POST", "/api/hosts/tls-probe", True),
("deploy", "PATCH", "/api/hosts/abc", True),
("deploy", "DELETE", "/api/hosts/abc", True),
("deploy", "POST", "/api/hosts/abc/maintenance/execute", False),
("deploy", "POST", "/api/users", True),
("deploy", "PUT", "/api/settings", True),
("deploy", "PUT", "/api/notifications/settings", True),
("deploy", "PUT", "/api/logs/settings", True),
("deploy", "POST", "/api/system/updater/update", True),
("admin", "POST", "/api/users", False),
("admin", "POST", "/api/hosts", False),
# Session-only for every scope, reads included.
("admin", "GET", "/api/tokens", True),
("admin", "POST", "/api/tokens", True),
("read", "GET", "/api/tokens", True),
("admin", "POST", "/api/auth/change-password", True),
("admin", "POST", "/api/auth/2fa/disable", True),
],
)
def test_scope_matrix(scope, method, path, denied):
assert api_token_service.denies(scope, method, path) is denied
@pytest.mark.asyncio
async def test_create_needs_password_and_shows_token_once(client):
bad = await client.post("/api/tokens", json={"name": "x", "current_password": "wrong"})
assert bad.status_code == 401
assert bad.json()["detail"]["code"] == "AKTUELLES_PASSWORT_IST_FALSCH"
created = await _make(client, expires_in_days=30)
assert created["token"].startswith("opx_")
assert created["scope"] == "read"
assert created["prefix"] == created["token"][:10]
assert created["expires_at"] is not None
listed = (await client.get("/api/tokens")).json()
assert [t["id"] for t in listed] == [created["id"]]
assert "token" not in listed[0] and created["token"] not in str(listed)
@pytest.mark.asyncio
async def test_only_the_hash_is_stored(client):
created = await _make(client)
async with SessionLocal() as db:
row = await db.scalar(select(ApiToken))
assert row.token_hash == api_token_service.hash_token(created["token"])
assert created["token"] not in (row.token_hash, row.prefix)
@pytest.mark.asyncio
async def test_bearer_authenticates_without_a_session(client, anon_client):
created = await _make(client)
async with _bearer(created["token"]) as api:
assert (await api.get("/api/hosts")).status_code == 200
me = (await api.get("/api/auth/me")).json()
assert me["authenticated"] and me["via"] == "token" and me["scope"] == "read"
assert (await anon_client.get("/api/hosts")).status_code == 401
@pytest.mark.asyncio
async def test_read_scope_cannot_write(client):
created = await _make(client, "read")
async with _bearer(created["token"]) as api:
resp = await api.post("/api/groups", json={"name": "g"})
assert resp.status_code == 403
assert resp.json()["detail"]["code"] == "API_TOKEN_SCOPE_FORBIDDEN"
@pytest.mark.asyncio
async def test_deploy_scope_writes_fleet_objects_but_not_access_control(client):
created = await _make(client, "deploy")
async with _bearer(created["token"]) as api:
assert (await api.post("/api/groups", json={"name": "from-ci"})).status_code == 201
for method, path, body in [
("POST", "/api/users", {"username": "evil", "password": "evil-password-123", "role": "admin"}),
("POST", "/api/hosts", {}),
]:
resp = await api.request(method, path, json=body)
assert resp.status_code == 403, (path, resp.text)
assert resp.json()["detail"]["code"] == "API_TOKEN_SCOPE_FORBIDDEN"
@pytest.mark.asyncio
async def test_tokens_cannot_manage_tokens_or_accounts(client):
created = await _make(client, "admin")
async with _bearer(created["token"]) as api:
for method, path in [("GET", "/api/tokens"), ("POST", "/api/tokens"), ("DELETE", f"/api/tokens/{created['id']}"),
("POST", "/api/auth/change-password")]:
resp = await api.request(method, path, json={"name": "x", "current_password": PASSWORD})
assert resp.status_code == 403, (path, resp.text)
assert resp.json()["detail"]["code"] == "API_TOKEN_SCOPE_FORBIDDEN"
@pytest.mark.asyncio
async def test_admin_scope_acts_as_admin_and_is_audited_with_token_name(client):
created = await _make(client, "admin", name="terraform")
async with _bearer(created["token"]) as api:
resp = await api.post("/api/users", json={"username": "viaapi", "password": "via-api-password-1", "role": "read_only"})
assert resp.status_code == 201, resp.text
items = (await client.get("/api/audit", params={"kind": "user"})).json()["items"]
assert any(e["actor"] == "admin (API: terraform)" for e in items), items
@pytest.mark.asyncio
async def test_bad_or_expired_token_is_401_and_never_falls_back_to_the_cookie(client):
# `client` carries a valid session cookie; a broken bearer header must still fail.
resp = await client.get("/api/hosts", headers={"Authorization": "Bearer opx_doesnotexist"})
assert resp.status_code == 401
created = await _make(client)
async with SessionLocal() as db:
await db.execute(update(ApiToken).values(expires_at=datetime.now(timezone.utc) - timedelta(seconds=1)))
await db.commit()
async with _bearer(created["token"]) as api:
assert (await api.get("/api/hosts")).status_code == 401
@pytest.mark.asyncio
async def test_non_token_bearer_values_are_ignored(client):
# Only opx_ values are API tokens; anything else leaves normal cookie auth alone.
resp = await client.get("/api/hosts", headers={"Authorization": "Bearer something-else"})
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_revoked_token_stops_working(client):
created = await _make(client)
async with _bearer(created["token"]) as api:
assert (await api.get("/api/hosts")).status_code == 200
assert (await client.delete(f"/api/tokens/{created['id']}")).status_code == 204
assert (await api.get("/api/hosts")).status_code == 401
assert (await client.delete(f"/api/tokens/{created['id']}")).status_code == 404
@pytest.mark.asyncio
async def test_last_used_is_recorded(client):
created = await _make(client)
async with _bearer(created["token"]) as api:
await api.get("/api/hosts")
listed = (await client.get("/api/tokens")).json()[0]
assert listed["last_used_at"] is not None
@pytest.mark.asyncio
async def test_read_only_user_gets_only_read_tokens_and_sees_only_their_own(client):
resp = await client.post("/api/users", json={"username": "tokviewer", "password": "viewer-password-123", "role": "read_only"})
assert resp.status_code == 201, resp.text
admin_token = await _make(client, "admin", name="admins")
viewer = await _login("tokviewer", "viewer-password-123")
try:
denied = await viewer.post("/api/tokens", json={"name": "x", "scope": "deploy", "current_password": "viewer-password-123"})
assert denied.status_code == 403
assert denied.json()["detail"]["code"] == "API_TOKEN_SCOPE_TOO_HIGH"
ok = await viewer.post("/api/tokens", json={"name": "grafana", "scope": "read", "current_password": "viewer-password-123"})
assert ok.status_code == 201, ok.text
mine = (await viewer.get("/api/tokens")).json()
assert [t["name"] for t in mine] == ["grafana"]
# Not someone else's token.
assert (await viewer.delete(f"/api/tokens/{admin_token['id']}")).status_code == 404
finally:
await viewer.aclose()
# The admin sees both and may revoke the viewer's.
all_tokens = {t["name"]: t for t in (await client.get("/api/tokens")).json()}
assert set(all_tokens) == {"admins", "grafana"} and all_tokens["grafana"]["owner"] == "tokviewer"
assert (await client.delete(f"/api/tokens/{all_tokens['grafana']['id']}")).status_code == 204
@pytest.mark.asyncio
async def test_demoted_owner_makes_the_token_read_only(client):
users = await client.post("/api/users", json={"username": "demotee", "password": "demotee-password-1", "role": "admin"})
uid = users.json()["id"]
owner = await _login("demotee", "demotee-password-1")
try:
created = (await owner.post("/api/tokens", json={"name": "t", "scope": "admin", "current_password": "demotee-password-1"})).json()
finally:
await owner.aclose()
assert (await client.patch(f"/api/users/{uid}/role", json={"role": "read_only"})).status_code == 200
async with _bearer(created["token"]) as api:
resp = await api.post("/api/groups", json={"name": "g"})
assert resp.status_code == 403
assert resp.json()["detail"]["code"] == "READ_ONLY_ACCESS"
@pytest.mark.asyncio
async def test_password_reset_and_user_delete_revoke_tokens(client):
resp = await client.post("/api/users", json={"username": "resetme", "password": "resetme-password-1", "role": "admin"})
uid = resp.json()["id"]
owner = await _login("resetme", "resetme-password-1")
try:
t1 = (await owner.post("/api/tokens", json={"name": "a", "current_password": "resetme-password-1"})).json()
finally:
await owner.aclose()
async with _bearer(t1["token"]) as api:
assert (await api.get("/api/hosts")).status_code == 200
assert (await client.post(f"/api/users/{uid}/reset-password", json={"new_password": "brand-new-password-1"})).status_code == 200
assert (await api.get("/api/hosts")).status_code == 401
assert (await client.delete(f"/api/users/{uid}")).status_code == 204
async with SessionLocal() as db:
assert (await db.scalars(select(ApiToken))).all() == []
@pytest.mark.asyncio
async def test_token_count_is_limited(client, monkeypatch):
monkeypatch.setattr(api_token_service, "MAX_TOKENS_PER_USER", 2)
await _make(client, name="1")
await _make(client, name="2")
resp = await client.post("/api/tokens", json={"name": "3", "current_password": PASSWORD})
assert resp.status_code == 409
assert resp.json()["detail"]["code"] == "API_TOKEN_LIMIT_REACHED"
@pytest.mark.asyncio
async def test_wrong_password_attempts_are_throttled(client):
codes = []
for _ in range(15):
codes.append((await client.post("/api/tokens", json={"name": "x", "current_password": "nope"})).status_code)
assert 429 in codes
@pytest.mark.asyncio
async def test_openapi_schema_is_served_behind_auth_with_bearer_scheme(client, anon_client):
assert (await anon_client.get("/api/openapi.json")).status_code == 401
schema = (await client.get("/api/openapi.json")).json()
assert schema["components"]["securitySchemes"]["bearerAuth"]["scheme"] == "bearer"
assert "/api/tokens" in schema["paths"] and "/api/hosts" in schema["paths"]
created = await _make(client)
async with _bearer(created["token"]) as api:
assert (await api.get("/api/openapi.json")).status_code == 200
# No interactive docs page (its CDN scripts would be blocked by the CSP anyway).
assert (await client.get("/docs")).status_code != 200
+6 -4
View File
@@ -92,13 +92,15 @@ async def test_health_stays_public_but_docs_require_auth(anon_client):
# /docs/redoc/openapi.json used to be public — now require a session
# like any other endpoint, since fully documenting a firewall-fleet
# API surface to an anonymous caller is free recon for an attacker.
for path in ("/docs", "/redoc", "/openapi.json"):
for path in ("/docs", "/redoc", "/openapi.json", "/api/openapi.json"):
resp = await anon_client.get(path)
assert resp.status_code == 401
@pytest.mark.asyncio
async def test_docs_reachable_once_logged_in(client):
async def test_openapi_schema_reachable_once_logged_in(client):
# The schema lives under /api (the proxy only routes /api to the backend); the
# interactive Swagger/ReDoc pages are gone (their CDN scripts violate the CSP).
assert (await client.get("/api/openapi.json")).status_code == 200
for path in ("/docs", "/redoc", "/openapi.json"):
resp = await client.get(path)
assert resp.status_code == 200
assert (await client.get(path)).status_code == 404
+214
View File
@@ -0,0 +1,214 @@
"""Security posture: check logic on config.xml shapes taken from real OPNsense 26.7 installs,
the stored-result API, and the metrics that follow from it."""
from datetime import datetime, timedelta, timezone
import httpx
import pytest
from app.services import compliance_service as cs
from tests.test_api_tokens import _bearer, _make
from tests.test_rules import _create_host
def _mvc_rule(*, interface="lan", action="pass", src="any", dst="any", protocol="any", descr="r", seq=1,
enabled="1", dport="", direction="in", ipprotocol="inet", quick="1", src_not="0"):
return (f"<rule><enabled>{enabled}</enabled><action>{action}</action><quick>{quick}</quick><interface>{interface}</interface>"
f"<direction>{direction}</direction><ipprotocol>{ipprotocol}</ipprotocol><protocol>{protocol}</protocol>"
f"<source_net>{src}</source_net><source_not>{src_not}</source_not><destination_net>{dst}</destination_net>"
f"<destination_not>0</destination_not><destination_port>{dport}</destination_port>"
f"<description>{descr}</description><sequence>{seq}</sequence></rule>")
def _config(*, rules="", legacy="", protocol="https", ssh="", snmpd="", timeservers="0.pool.ntp.org", syslog="",
interfaces='<wan><if>vtnet0</if></wan><lan><if>vtnet1</if></lan>', ids="0"):
return (f"<opnsense><system><webgui><protocol>{protocol}</protocol></webgui>"
f"<timeservers>{timeservers}</timeservers>{ssh}</system><interfaces>{interfaces}</interfaces>"
f"<filter>{legacy}</filter>{snmpd}<OPNsense><Firewall><Filter><rules>{rules}</rules></Filter></Firewall>"
f"<IDS><general><enabled>{ids}</enabled></general></IDS><Syslog><destinations>{syslog}</destinations></Syslog>"
f"</OPNsense></opnsense>").encode()
def _run(xml, firmware="none", backup_age=3600.0):
return {f["id"]: f for f in cs.evaluate(xml, firmware_status=firmware, newest_backup_age_s=backup_age)}
def test_clean_config_passes_everything_and_scores_100():
findings = cs.evaluate(
_config(rules=_mvc_rule(descr="lan out", src="lan"), syslog="<destination><enabled>1</enabled></destination>", ids="1"),
firmware_status="none", newest_backup_age_s=3600,
)
assert [f["id"] for f in findings if f["passed"] is False] == []
assert cs.score(findings) == 100 and cs.grade(100) == "good"
def test_wan_any_any_is_critical_and_names_the_rule():
# What the OPNexus test firewalls have: a pass-everything rule on the WAN.
by_id = _run(_config(rules=_mvc_rule(interface="wan", descr="Allow ALL")))
assert by_id["wan_any_any"]["passed"] is False and by_id["wan_any_any"]["severity"] == "critical"
assert by_id["wan_any_any"]["params"] == {"count": 1, "examples": ["Allow ALL"]}
assert cs.score(list(by_id.values())) <= 70
def test_default_lan_rule_is_not_a_finding_but_a_lan_any_source_is():
assert _run(_config(rules=_mvc_rule(src="lan", descr="Default allow LAN to any rule")))["internal_any_any"]["passed"] is True
assert _run(_config(rules=_mvc_rule(src="any", descr="open")))["internal_any_any"]["passed"] is False
def test_a_wan_rule_with_a_port_or_protocol_is_not_any_any():
for extra in ({"dport": "443"}, {"protocol": "TCP"}):
assert _run(_config(rules=_mvc_rule(interface="wan", **extra)))["wan_any_any"]["passed"] is True
def test_disabled_rules_do_not_count():
assert _run(_config(rules=_mvc_rule(interface="wan", enabled="0")))["wan_any_any"]["passed"] is True
def test_legacy_rule_format_is_understood():
legacy = ("<rule><type>pass</type><interface>wan</interface><ipprotocol>inet</ipprotocol><descr>ALLOW ANY</descr>"
"<source><any>1</any></source><destination><any/></destination></rule>")
by_id = _run(_config(legacy=legacy))
assert by_id["wan_any_any"]["passed"] is False and by_id["wan_any_any"]["params"]["examples"] == ["ALLOW ANY"]
def test_interface_with_a_gateway_counts_as_wan():
interfaces = "<opt1><if>vtnet2</if><gateway>GW1</gateway></opt1><lan><if>vtnet1</if></lan>"
assert _run(_config(interfaces=interfaces, rules=_mvc_rule(interface="opt1")))["wan_any_any"]["passed"] is False
def test_shadowed_and_duplicate_rules():
rules = (
_mvc_rule(interface="lan", src="any", dst="any", action="block", descr="block all", seq=1)
+ _mvc_rule(interface="lan", src="10.0.0.0/8", dst="any", action="pass", descr="never reached", seq=2)
+ _mvc_rule(interface="dmz", src="dmz", dst="any", descr="a", seq=3)
+ _mvc_rule(interface="dmz", src="dmz", dst="any", descr="a again", seq=4)
)
by_id = _run(_config(rules=rules))
assert by_id["rules_shadowed"]["passed"] is False and by_id["rules_shadowed"]["params"]["examples"] == ["never reached"]
assert by_id["rules_duplicate"]["passed"] is False and by_id["rules_duplicate"]["params"]["examples"] == ["a again"]
def test_coverage_is_never_claimed_without_proof():
rules = (
_mvc_rule(interface="lan", src="lan", dst="any", dport="443", seq=1, descr="https only")
+ _mvc_rule(interface="lan", src="lan", dst="any", dport="80", seq=2, descr="http")
+ _mvc_rule(interface="lan", src="lan", dst="any", seq=3, descr="everything")
+ _mvc_rule(interface="lan", src="lan", dst="any", ipprotocol="inet6", seq=4, descr="v6")
+ _mvc_rule(interface="lan", src="lan", dst="any", src_not="1", seq=5, descr="negated")
)
by_id = _run(_config(rules=rules))
assert by_id["rules_shadowed"]["passed"] is True and by_id["rules_duplicate"]["passed"] is True
def test_rules_are_ordered_by_sequence_not_file_order():
rules = _mvc_rule(src="10.0.0.0/8", descr="narrow", seq=9) + _mvc_rule(src="any", descr="broad", seq=1, action="block")
assert _run(_config(rules=rules))["rules_shadowed"]["params"]["examples"] == ["narrow"]
def test_ssh_levels():
on = "<ssh><enabled>enabled</enabled><permitrootlogin>{}</permitrootlogin><passwordauth>{}</passwordauth></ssh>"
worst = _run(_config(ssh=on.format(1, 1)))["ssh_hardened"]
assert (worst["passed"], worst["severity"]) == (False, "high")
mid = _run(_config(ssh=on.format(0, 1)))["ssh_hardened"]
assert (mid["passed"], mid["severity"]) == (False, "medium")
assert _run(_config(ssh=on.format(0, 0)))["ssh_hardened"]["passed"] is True
assert _run(_config(ssh="<ssh><permitrootlogin>1</permitrootlogin><passwordauth>1</passwordauth></ssh>"))["ssh_hardened"]["passed"] is True
assert _run(_config())["ssh_hardened"]["passed"] is True
def test_webgui_snmp_ntp_logging_ids():
assert _run(_config(protocol="http"))["webgui_https"]["passed"] is False
assert "snmp_community" not in _run(_config())
assert _run(_config(snmpd="<snmpd><rocommunity>public</rocommunity></snmpd>"))["snmp_community"]["passed"] is False
assert _run(_config(snmpd="<snmpd><rocommunity>s3cret-ro</rocommunity></snmpd>"))["snmp_community"]["passed"] is True
assert _run(_config(timeservers=""))["ntp_configured"]["passed"] is False
assert _run(_config())["remote_logging"]["passed"] is False
assert _run(_config(ids="0"))["ids_enabled"]["severity"] == "info"
def test_firmware_and_backup_inputs():
assert _run(_config(), firmware="update")["firmware_current"]["passed"] is False
assert _run(_config(), firmware=None)["firmware_current"]["passed"] is None
assert _run(_config(), backup_age=None)["backup_recent"]["passed"] is False
assert _run(_config(), backup_age=8 * 86400)["backup_recent"]["passed"] is False
assert _run(_config(), backup_age=2 * 86400)["backup_recent"]["passed"] is True
def test_unknown_results_do_not_cost_points_and_info_is_free():
findings = cs.evaluate(_config(), firmware_status=None, newest_backup_age_s=0)
failed = [f["id"] for f in findings if f["passed"] is False]
assert cs.score(findings) == 100 - sum(cs.PENALTY[f["severity"]] for f in findings if f["passed"] is False)
assert "firmware_current" not in failed and "ids_enabled" in failed and cs.PENALTY["info"] == 0
@pytest.mark.parametrize("bad", [b"", b" ", b"<html/>", b"<opnsense>", b'<!DOCTYPE x [<!ENTITY a "b">]><opnsense>&a;</opnsense>'])
def test_invalid_or_hostile_input_is_rejected(bad):
with pytest.raises(Exception):
cs.evaluate(bad, firmware_status=None, newest_backup_age_s=None)
def test_findings_carry_no_config_secrets():
xml = _config(rules=_mvc_rule(interface="wan", descr="Allow ALL")).replace(
b"<opnsense>", b"<opnsense><system2><user><password>$2y$hash</password><apikey>KEY123</apikey></user></system2>")
assert b"KEY123" not in repr(cs.evaluate(xml, firmware_status=None, newest_backup_age_s=None)).encode()
# ── API ──────────────────────────────────────────────────────────────────
def _mock_host(respx_mock, xml, firmware_status="none"):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
respx_mock.get(url__regex=r".*/core/backup/download/this").mock(return_value=httpx.Response(200, content=xml))
respx_mock.get(url__regex=r".*/core/firmware/status").mock(return_value=httpx.Response(200, json={"status": firmware_status}))
@pytest.mark.asyncio
async def test_check_stores_and_lists_the_result(client, respx_mock):
_mock_host(respx_mock, _config(rules=_mvc_rule(interface="wan", descr="Allow ALL")))
host = await _create_host(client, "fw1", "https://fw1.test")
assert (await client.get(f"/api/hosts/{host['id']}/compliance")).json()["checked_at"] is None
result = (await client.post(f"/api/hosts/{host['id']}/compliance/check")).json()
assert result["error"] is None and result["score"] < 100 and result["grade"] in ("fair", "poor")
assert result["counts"]["critical"] == 1
assert next(f for f in result["findings"] if f["id"] == "wan_any_any")["passed"] is False
stored = (await client.get(f"/api/hosts/{host['id']}/compliance")).json()
assert stored["score"] == result["score"] and stored["findings"] == result["findings"]
fleet = (await client.get("/api/compliance")).json()
assert [(h["host_name"], h["score"]) for h in fleet] == [("fw1", result["score"])] and "findings" not in fleet[0]
# Second check replaces the row instead of adding one.
await client.post(f"/api/hosts/{host['id']}/compliance/check")
assert len((await client.get("/api/compliance")).json()) == 1
@pytest.mark.asyncio
async def test_unreachable_host_is_reported_not_scored(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
host = await _create_host(client, "fw1", "https://fw1.test")
respx_mock.get(url__regex=r".*/core/backup/download/this").mock(return_value=httpx.Response(500))
result = (await client.post(f"/api/hosts/{host['id']}/compliance/check")).json()
assert result["error"] == "unreachable" and result["score"] is None and result["grade"] is None
@pytest.mark.asyncio
async def test_garbage_config_is_reported_as_invalid(client, respx_mock):
_mock_host(respx_mock, b"<html>not a config</html>")
host = await _create_host(client, "fw1", "https://fw1.test")
assert (await client.post(f"/api/hosts/{host['id']}/compliance/check")).json()["error"] == "invalid_config"
@pytest.mark.asyncio
async def test_unknown_host_is_404_and_checks_need_write_rights(client):
import uuid
assert (await client.get(f"/api/hosts/{uuid.uuid4()}/compliance")).status_code == 404
token = await _make(client, "read")
async with _bearer(token["token"]) as api:
assert (await api.get("/api/compliance")).status_code == 200 # reading is fine
assert (await api.post("/api/compliance/check")).status_code == 403 # running a check is not
@pytest.mark.asyncio
async def test_metrics_expose_the_score(client, respx_mock):
_mock_host(respx_mock, _config(rules=_mvc_rule(interface="wan", descr="Allow ALL")))
host = await _create_host(client, "fw1", "https://fw1.test")
result = (await client.post(f"/api/hosts/{host['id']}/compliance/check")).json()
text = (await client.get("/api/metrics")).text
assert f'opnexus_host_compliance_score{{host="fw1",cluster=""}} {result["score"]}' in text
assert 'opnexus_host_compliance_findings{severity="critical",host="fw1",cluster=""} 1' in text
+105
View File
@@ -0,0 +1,105 @@
"""Read-only MCP endpoint: protocol basics, tools, and that it cannot be used to escalate."""
import pytest
from tests.test_api_tokens import _bearer, _make
def _rpc(method, params=None, id_=1):
message = {"jsonrpc": "2.0", "method": method}
if id_ is not None:
message["id"] = id_
if params is not None:
message["params"] = params
return message
async def _call(api, name, **arguments):
resp = await api.post("/api/mcp", json=_rpc("tools/call", {"name": name, "arguments": arguments}))
assert resp.status_code == 200, resp.text
return resp.json()["result"]
@pytest.mark.asyncio
async def test_requires_authentication(anon_client):
assert (await anon_client.post("/api/mcp", json=_rpc("ping"))).status_code == 401
@pytest.mark.asyncio
async def test_handshake_and_tool_list_with_a_read_token(client):
token = await _make(client, "read")
async with _bearer(token["token"]) as api:
init = (await api.post("/api/mcp", json=_rpc("initialize", {"protocolVersion": "2025-06-18"}))).json()["result"]
assert init["protocolVersion"] == "2025-06-18"
assert init["serverInfo"]["name"] == "opnexus" and "tools" in init["capabilities"]
# Unknown future version: answer with one we speak.
other = (await api.post("/api/mcp", json=_rpc("initialize", {"protocolVersion": "2099-01-01"}))).json()["result"]
assert other["protocolVersion"] == "2025-06-18"
# A notification gets no body.
note = await api.post("/api/mcp", json=_rpc("notifications/initialized", id_=None))
assert note.status_code == 202 and note.content == b""
tools = (await api.post("/api/mcp", json=_rpc("tools/list"))).json()["result"]["tools"]
assert {"fleet_overview", "list_hosts", "recent_audit", "search", "api_get"} <= {t["name"] for t in tools}
assert all("inputSchema" in t and "path" not in t for t in tools)
assert (await api.post("/api/mcp", json=_rpc("ping"))).json()["result"] == {}
unknown = (await api.post("/api/mcp", json=_rpc("resources/list"))).json()
assert unknown["error"]["code"] == -32601
@pytest.mark.asyncio
async def test_get_is_405_and_bad_json_is_a_parse_error(client):
assert (await client.get("/api/mcp")).status_code == 405
bad = await client.post("/api/mcp", content=b"{nope", headers={"Content-Type": "application/json"})
assert bad.status_code == 400 and bad.json()["error"]["code"] == -32700
@pytest.mark.asyncio
async def test_tools_read_through_the_callers_own_permissions(client):
await client.post("/api/groups", json={"name": "findable-group"})
token = await _make(client, "read")
async with _bearer(token["token"]) as api:
hosts = await _call(api, "list_hosts")
assert hosts["isError"] is False and hosts["content"][0]["text"] == "[]"
found = await _call(api, "search", q="findable")
assert "findable-group" in found["content"][0]["text"]
audit = await _call(api, "recent_audit", limit=5)
assert audit["isError"] is False
overview = await _call(api, "fleet_overview")
assert "pending_actions" in overview["content"][0]["text"]
@pytest.mark.asyncio
async def test_generic_reader_cannot_reach_credentials_or_secrets_and_never_writes(client):
token = await _make(client, "read")
async with _bearer(token["token"]) as api:
for path in ("/api/tokens", "/api/auth/me", "/api/mcp", "/etc/passwd", "/api/../x", "/api/hosts?x=1", "relative"):
result = await _call(api, "api_get", path=path)
assert result["isError"] is True, path
# Secret-bearing endpoint: refused by the caller's own (read) scope, not by the tool.
denied = await _call(api, "api_get", path="/api/notifications/settings")
assert denied["isError"] is True and "API_TOKEN_SCOPE_FORBIDDEN" in denied["content"][0]["text"]
# api_get is GET-only by construction: there is no way to name another method.
tools = (await api.post("/api/mcp", json=_rpc("tools/list"))).json()["result"]["tools"]
schema = next(t for t in tools if t["name"] == "api_get")["inputSchema"]
assert set(schema["properties"]) == {"path", "query"}
@pytest.mark.asyncio
async def test_unknown_tool_and_bad_arguments_are_tool_errors(client):
token = await _make(client, "read")
async with _bearer(token["token"]) as api:
assert (await _call(api, "delete_everything"))["isError"] is True
assert (await _call(api, "host_resources"))["isError"] is True
bad = await api.post("/api/mcp", json=_rpc("tools/call", {"arguments": {}}))
assert bad.json()["error"]["code"] == -32602
@pytest.mark.asyncio
async def test_read_only_users_session_can_use_it_too(client):
resp = await client.post("/api/users", json={"username": "mcpviewer", "password": "viewer-password-123", "role": "read_only"})
assert resp.status_code == 201
from tests.test_api_tokens import _login
viewer = await _login("mcpviewer", "viewer-password-123")
try:
assert (await viewer.post("/api/mcp", json=_rpc("ping"))).status_code == 200
finally:
await viewer.aclose()
+104
View File
@@ -0,0 +1,104 @@
"""Prometheus endpoint: format, content and that it never calls a firewall."""
import re
from datetime import datetime, timedelta, timezone
import httpx
import pytest
from app.database import SessionLocal
from app.models.carp_event import CarpEvent
from app.models.notification import NotificationProblem
from app.models.resource_sample import HostResourceSample
from tests.test_api_tokens import _bearer, _make
from tests.test_rules import _create_host
_SAMPLE_LINE = re.compile(r'^[a-zA-Z_:][a-zA-Z0-9_:]*(\{([a-zA-Z_][a-zA-Z0-9_]*="([^"\\]|\\.)*"(,|(?=\})))*\})? -?\d+(\.\d+)?(e[+-]?\d+)?$')
def _parse(text: str) -> dict[str, float]:
"""name{labels} -> value, asserting the exposition grammar line by line."""
values = {}
for line in text.splitlines():
if not line or line.startswith("# HELP ") or line.startswith("# TYPE "):
continue
assert _SAMPLE_LINE.match(line), f"not valid exposition: {line!r}"
key, _, value = line.rpartition(" ")
values[key] = float(value)
return values
async def _seed(host_id: str, *, age_s: int, carp: str | None):
async with SessionLocal() as db:
db.add(HostResourceSample(
host_id=host_id, sampled_at=datetime.now(timezone.utc) - timedelta(seconds=age_s),
mem_used_pct=25.0, load_avg_1m=0.42, disk_used_pct=12.0, raw={},
))
if carp:
db.add(CarpEvent(host_id=host_id, host_name="fw1", cluster_name="ha", from_status=None, to_status=carp))
await db.commit()
@pytest.mark.asyncio
async def test_metrics_requires_auth_and_is_prometheus_text(client, anon_client):
assert (await anon_client.get("/api/metrics")).status_code == 401
resp = await client.get("/api/metrics")
assert resp.status_code == 200
assert resp.headers["content-type"].startswith("text/plain; version=0.0.4")
values = _parse(resp.text)
assert values["opnexus_hosts"] == 0
assert any(key.startswith('opnexus_info{version="') for key in values)
@pytest.mark.asyncio
async def test_host_metrics_from_stored_samples_without_calling_the_firewall(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
host = await _create_host(client, "fw1", "https://fw1.test")
await _seed(host["id"], age_s=30, carp="MASTER")
before = len(respx_mock.calls)
values = _parse((await client.get("/api/metrics")).text)
# The scrape is DB-only: not a single request went to a firewall.
assert len(respx_mock.calls) == before
assert values["opnexus_hosts"] == 1
assert values['opnexus_host_up{host="fw1",cluster=""}'] == 1
assert values['opnexus_host_memory_used_percent{host="fw1",cluster=""}'] == 25
assert values['opnexus_host_load1{host="fw1",cluster=""}'] == 0.42
assert values['opnexus_host_disk_used_percent{host="fw1",cluster=""}'] == 12
assert values['opnexus_host_carp_master{host="fw1",cluster=""}'] == 1
@pytest.mark.asyncio
async def test_stale_sample_means_host_down_and_unknown_carp_is_omitted(client):
host = await _create_host(client, "fw1", "https://fw1.test")
await _seed(host["id"], age_s=600, carp=None)
text = (await client.get("/api/metrics")).text
values = _parse(text)
assert values['opnexus_host_up{host="fw1",cluster=""}'] == 0
assert "opnexus_host_carp_master{" not in text
@pytest.mark.asyncio
async def test_open_problems_are_counted_and_resolved_ones_are_not(client):
async with SessionLocal() as db:
db.add(NotificationProblem(problem_type="host_offline", dedup_key="a", status="open"))
db.add(NotificationProblem(problem_type="host_offline", dedup_key="b", status="open"))
db.add(NotificationProblem(problem_type="cert_expiry", dedup_key="c", status="resolved"))
await db.commit()
values = _parse((await client.get("/api/metrics")).text)
assert values['opnexus_problems_open{type="host_offline"}'] == 2
assert not any("cert_expiry" in key for key in values)
@pytest.mark.asyncio
async def test_label_values_are_escaped(client):
await _create_host(client, 'we"ird\\name', "https://fw1.test")
text = (await client.get("/api/metrics")).text
assert 'host="we\\"ird\\\\name"' in text
_parse(text)
@pytest.mark.asyncio
async def test_read_token_can_scrape(client):
created = await _make(client, "read", name="prometheus")
async with _bearer(created["token"]) as api:
resp = await api.get("/api/metrics")
assert resp.status_code == 200 and "opnexus_info" in resp.text
+112 -6
View File
@@ -26,12 +26,13 @@ Eine Kurzfassung steht in der [README](../README.md), die Änderungen je Version
12. [HA-Cluster und CARP](#ha-cluster-und-carp)
13. [Wartung und Updates der Firewalls](#wartung-und-updates-der-firewalls)
14. [Überwachung: Dashboard, Logs, Audit, Benachrichtigungen](#überwachung-dashboard-logs-audit-benachrichtigungen)
15. [OPNexus betreiben](#opnexus-betreiben)
16. [Sicherheit](#sicherheit)
17. [Fehlerbehebung](#fehlerbehebung)
18. [Grenzen](#grenzen)
19. [Bedienung, Sprache, Barrierefreiheit](#bedienung-sprache-barrierefreiheit)
20. [Lizenz und Datenschutz](#lizenz-und-datenschutz)
15. [Automatisierung: API-Tokens, Prometheus, KI-Assistenten](#automatisierung-api-tokens-prometheus-ki-assistenten)
16. [OPNexus betreiben](#opnexus-betreiben)
17. [Sicherheit](#sicherheit)
18. [Fehlerbehebung](#fehlerbehebung)
19. [Grenzen](#grenzen)
20. [Bedienung, Sprache, Barrierefreiheit](#bedienung-sprache-barrierefreiheit)
21. [Lizenz und Datenschutz](#lizenz-und-datenschutz)
---
@@ -782,6 +783,36 @@ ausgeführt“: Sie bleibt drei Tage nach dem geplanten Zeitpunkt offen und schl
ohne „behoben“-Meldung, weil sich ein verpasster Termin nicht von selbst behebt. Der
Text der Benachrichtigungen ist nicht übersetzt.
### Sicherheitslage
Unter **Monitoring › Sicherheitslage** prüft OPNexus jede Firewall auf typische
Schwachstellen und vergibt einen Wert von 0 bis 100 (ab 90 gut, ab 70 mittel, darunter
schlecht). Die Prüfung liest die Konfiguration der Firewall einmal lesend herunter
und wertet sie im Speicher aus; es wird nichts an der Firewall geändert und die
Konfiguration nicht gespeichert. Alle sechs Stunden läuft die Prüfung von selbst,
„Alle prüfen“ und „Erneut prüfen“ starten sie sofort.
| Prüfung | Gewicht | Wann ein Befund entsteht |
|---|---|---|
| Keine Allow-all-Regel am WAN | kritisch | Eine aktive Regel am WAN (oder an einer Schnittstelle mit Upstream-Gateway) erlaubt jeden Verkehr von überall nach überall |
| Web-Oberfläche nur über HTTPS | hoch | Die Verwaltungsoberfläche läuft über HTTP |
| SSH abgesichert | hoch / mittel | SSH ist an und erlaubt Passwort-Anmeldung (hoch, wenn auch root per Passwort darf) |
| SNMP-Community nicht „public“ | hoch | SNMP ist an und nutzt „public“ |
| Keine Allow-all-Regel mit beliebiger Quelle (intern) | mittel | Eine interne Regel erlaubt beliebige Quellen zu beliebigen Zielen (die Standardregel „LAN to any“ zählt nicht) |
| Keine verdeckten Regeln | mittel | Eine Regel wird nie erreicht, weil eine frühere sie schon abdeckt |
| Firmware aktuell | mittel | Updates stehen bereit |
| Aktuelles Backup | mittel | Das neueste lokale Backup ist 7 Tage oder älter |
| Keine doppelten Regeln, Regeln beschriftet, Zeitserver, externe Logs | niedrig | Doppelte Regel, Regel ohne Beschreibung, kein Zeitserver, kein Syslog-Ziel |
| Einbruchserkennung aktiv | Hinweis | IDS/IPS ist aus (kostet keine Punkte) |
Ein kritischer Befund kostet 30 Punkte, ein hoher 15, ein mittlerer 8, ein niedriger 3.
Nicht entscheidbare Prüfungen (etwa der Updatestatus einer gerade nicht erreichbaren
Firewall) kosten nichts. Verdeckt oder doppelt meldet OPNexus nur, was sich beweisen
lässt: Regeln werden Feld für Feld verglichen, Aliase werden nicht aufgelöst, also
bleibt eine Überschneidung über Aliase unentdeckt, aber es gibt keine falschen
Befunde. Beide Regelformate (OPNsense 26.x und ältere Installationen) werden gelesen.
Die Werte stehen auch als Prometheus-Metriken zur Verfügung (siehe unten).
### Über OPNexus
Ein Klick auf das Logo öffnet **About**: alle Hosts mit Status, CARP-Rolle, Cluster,
@@ -791,6 +822,81 @@ jeder Seite und in den Einstellungen.
---
## Automatisierung: API-Tokens, Prometheus, KI-Assistenten
Unter **System › API-Tokens** legst du persönliche Tokens an. Mit einem Token
sprechen Skripte, CI-Jobs, Prometheus und KI-Assistenten die OPNexus-API ohne
Anmeldung an:
```
curl -H "Authorization: Bearer opx_…" https://opnexus.example/api/hosts
```
Ein Token gehört deinem Konto und hat nie mehr Rechte als du. Es wird nur einmal
angezeigt und nur als Hash gespeichert. Zum Anlegen brauchst du dein Passwort. Standard
ist eine Laufzeit von 90 Tagen; du kannst jedes Token jederzeit widerrufen, und ein
Passwort-Reset des Kontos widerruft alle seine Tokens. Höchstens 50 Tokens je Benutzer.
| Bereich | Darf |
|---|---|
| **Lesen** | Alles ansehen (GET), außer Dinge mit Geheimnissen: Roh-Download der Konfigurations-Backups und die Benachrichtigungseinstellungen. Für Dashboards, Prometheus, KI-Assistenten. |
| **Ausrollen** | Zusätzlich ausrollen und warten: Regeln, Aliase, NAT, VPN, Zertifikate, Wartung, Sicherheitsprüfung. Nicht: Benutzer, Einstellungen, Benachrichtigungen, Hosts anlegen oder löschen, OPNexus-Update. |
| **Admin** | Alles, was dein Konto darf, einschließlich der Dinge oben. |
Für jeden Bereich gilt: Tokens verwalten und das eigene Konto (Passwort, Zwei-Faktor)
geht nur in einer normalen Anmeldung, nie mit einem Token. Ein Nur-Lese-Konto kann
nur „Lesen“-Tokens anlegen; wird das Konto zurückgestuft, sind alle seine Tokens
sofort nur noch lesend. Im Audit steht bei Änderungen per Token „Benutzer (API: Name)“.
**API-Beschreibung:** das OpenAPI-Schema liegt unter `/api/openapi.json` (mit
Anmeldung oder Token abrufbar) und lässt sich in Postman, einen Code-Generator oder
einen Assistenten laden. Eine interaktive Seite gibt es nicht. Fehler kommen als
`{"detail": {"code": "…", "params": {…}}}` mit stabilem `code`.
### Prometheus
`GET /api/metrics` liefert den Zustand im Prometheus-Textformat. Die Antwort stammt
nur aus der OPNexus-Datenbank (keine Anfragen an die Firewalls), ein häufiges Abfragen
belastet also nichts. Mit einem „Lesen“-Token:
```yaml
scrape_configs:
- job_name: opnexus
scheme: https
metrics_path: /api/metrics
authorization: { credentials: "opx_…" }
static_configs: [{ targets: ["opnexus.example"] }]
```
| Metrik | Bedeutung |
|---|---|
| `opnexus_info{version}` | Version, immer 1 |
| `opnexus_hosts` | Anzahl der Firewalls |
| `opnexus_host_up{host,cluster}` | 1, wenn der Host in den letzten 3 Minuten geantwortet hat |
| `opnexus_host_enabled`, `…_last_sample_timestamp_seconds` | Überwachung an; Zeit der letzten Messung |
| `opnexus_host_memory_used_percent`, `opnexus_host_load1`, `opnexus_host_disk_used_percent` | Letzte Messung |
| `opnexus_host_carp_master` | 1 = MASTER, 0 = BACKUP (fehlt, wenn unbekannt) |
| `opnexus_host_compliance_score`, `opnexus_host_compliance_findings{severity}` | Sicherheitslage |
| `opnexus_pending_actions{kind,status}` | Offene Aktionen wie auf dem Dashboard |
| `opnexus_problems_open{type}` | Offene Probleme der Benachrichtigungen |
| `opnexus_scheduled_maintenance_pending` | Noch nicht gelaufene geplante Wartungen |
### KI-Assistenten (MCP)
OPNexus spricht das Model Context Protocol, damit ein KI-Assistent die Flotte
ansehen kann, **nur lesend**: Er kann nichts ändern. Beispiel für Claude Code:
```
claude mcp add --transport http opnexus https://opnexus.example/api/mcp \
--header "Authorization: Bearer opx_…"
```
Die Werkzeuge: `fleet_overview`, `list_hosts`, `host_resources`, `carp_events`,
`recent_audit`, `search` und `api_get` (liest jeden anderen GET-Endpunkt). Alles läuft
mit den Rechten des Tokens: ein „Lesen“-Token sieht auch hier keine Geheimnisse.
---
## OPNexus betreiben
### Daten und Volumes
+181
View File
@@ -0,0 +1,181 @@
"use client";
import { InfoTip } from "@/components/ui/Tooltip";
import { Badge } from "@/components/ui/Badge";
import { Button } from "@/components/ui/Button";
import { Card } from "@/components/ui/Card";
import { EmptyState } from "@/components/ui/EmptyState";
import { Skeleton } from "@/components/ui/Skeleton";
import { useToast } from "@/components/ui/Toast";
import { api, ComplianceFinding, ComplianceHost, Severity } from "@/lib/api";
import { useLanguage } from "@/components/LanguageProvider";
import { useReadOnly } from "@/lib/useReadOnly";
import { CheckCircle2, CircleHelp, Info, RefreshCw, ShieldAlert, XCircle } from "lucide-react";
import { useEffect, useState } from "react";
const SEVERITY_TONE: Record<Severity, "danger" | "warning" | "info" | "neutral"> = {
critical: "danger", high: "danger", medium: "warning", low: "info", info: "neutral",
};
const SEVERITY_RANK: Record<Severity, number> = { critical: 0, high: 1, medium: 2, low: 3, info: 4 };
const GRADE_TONE = { good: "success", fair: "warning", poor: "danger" } as const;
function ScoreBadge({ host }: { host: ComplianceHost }) {
const { t } = useLanguage();
if (host.score === null || host.grade === null) return <Badge tone="neutral">{host.error ? t(`security.error.${host.error}`) : t("security.notChecked")}</Badge>;
return <Badge tone={GRADE_TONE[host.grade]}>{host.score} / 100</Badge>;
}
export default function SecurityPage() {
const { t, errorMessage, locale } = useLanguage();
const readOnly = useReadOnly();
const toast = useToast();
const [hosts, setHosts] = useState<ComplianceHost[] | null>(null);
const [selected, setSelected] = useState<string | null>(null);
const [detail, setDetail] = useState<ComplianceHost | null>(null);
const [busy, setBusy] = useState(false);
const [error, setError] = useState<string | null>(null);
async function load() {
try {
const list = await api.listCompliance();
setHosts(list);
setSelected((current) => current ?? [...list].filter((h) => h.score !== null).sort((a, b) => (a.score ?? 0) - (b.score ?? 0))[0]?.host_id ?? list[0]?.host_id ?? null);
setError(null);
} catch (e) {
setError(errorMessage(e, "security.loadFailed"));
}
}
useEffect(() => {
load();
}, []);
useEffect(() => {
if (!selected) { setDetail(null); return; }
let active = true;
api.getHostCompliance(selected).then((d) => active && setDetail(d)).catch((e) => active && setError(errorMessage(e, "security.loadFailed")));
return () => { active = false; };
}, [selected, hosts]);
async function runCheck(all: boolean) {
setBusy(true);
try {
if (all || !selected) await api.checkFleetCompliance();
else await api.checkHostCompliance(selected);
await load();
toast.show(t("security.checked"), "success");
} catch (e) {
setError(errorMessage(e, "security.checkFailed"));
} finally {
setBusy(false);
}
}
const findings = [...(detail?.findings ?? [])].sort(
(a, b) => Number(a.passed === true) - Number(b.passed === true) || SEVERITY_RANK[a.severity] - SEVERITY_RANK[b.severity],
);
return (
<div>
<div className="mb-6 flex items-center justify-between gap-3">
<h1 className="flex items-center gap-2 text-2xl font-semibold text-slate-900">
{t("security.title")}
<InfoTip text={t("security.intro")} size={18} />
</h1>
<fieldset disabled={readOnly} className="contents">
<Button onClick={() => runCheck(true)} disabled={busy} loading={busy}>
<RefreshCw size={16} />
{t("security.checkAll")}
</Button>
</fieldset>
</div>
{error && <p className="mb-4 text-sm text-red-600">{error}</p>}
{hosts === null && <Skeleton className="h-24 w-full" />}
{hosts?.length === 0 && <EmptyState icon={ShieldAlert} title={t("security.noHosts")} description={t("security.noHostsHint")} />}
{hosts && hosts.length > 0 && (
<div className="mb-6 grid gap-3 sm:grid-cols-2 lg:grid-cols-3">
{hosts.map((host) => (
<button
key={host.host_id}
type="button"
onClick={() => setSelected(host.host_id)}
aria-pressed={selected === host.host_id}
className={`rounded-2xl border bg-white p-4 text-left transition-colors hover:border-slate-400 ${selected === host.host_id ? "border-slate-900 ring-1 ring-slate-900" : "border-slate-200"}`}
>
<div className="flex items-center justify-between gap-2">
<span className="font-medium text-slate-900">{host.host_name}</span>
<ScoreBadge host={host} />
</div>
<div className="mt-2 flex flex-wrap gap-1.5 text-xs">
{host.counts && (["critical", "high", "medium", "low"] as Severity[]).filter((s) => host.counts![s] > 0).map((s) => (
<Badge key={s} tone={SEVERITY_TONE[s]}>{host.counts![s]} {t(`security.severity.${s}`)}</Badge>
))}
{host.counts && Object.entries(host.counts).every(([k, v]) => k === "info" || v === 0) && host.score !== null && (
<span className="text-slate-500">{t("security.allGood")}</span>
)}
</div>
<p className="mt-2 text-xs text-slate-400">
{host.checked_at ? t("security.checkedAt", { when: new Date(host.checked_at).toLocaleString(locale) }) : t("security.neverChecked")}
</p>
</button>
))}
</div>
)}
{detail && (
<Card className="p-0">
<div className="flex items-center justify-between gap-3 border-b border-slate-100 px-5 py-3">
<h2 className="font-medium text-slate-900">{detail.host_name}</h2>
<fieldset disabled={readOnly} className="contents">
<Button variant="outline" onClick={() => runCheck(false)} disabled={busy}>
<RefreshCw size={14} />
{t("security.checkHost")}
</Button>
</fieldset>
</div>
{detail.error && <p className="px-5 py-4 text-sm text-red-600">{t(`security.errorHint.${detail.error}`)}</p>}
{!detail.error && findings.length === 0 && <p className="px-5 py-6 text-sm text-slate-500">{t("security.neverChecked")}</p>}
<ul className="divide-y divide-slate-100">
{findings.map((finding) => (
<FindingRow key={finding.id} finding={finding} />
))}
</ul>
</Card>
)}
</div>
);
}
function FindingRow({ finding }: { finding: ComplianceFinding }) {
const { t } = useLanguage();
const failed = finding.passed === false;
// A note (severity "info") is a suggestion, not an error: neutral icon instead of the red cross.
const note = failed && finding.severity === "info";
const Icon = finding.passed === null ? CircleHelp : note ? Info : failed ? XCircle : CheckCircle2;
const color = finding.passed === null || note ? "text-slate-400" : failed ? "text-red-600" : "text-emerald-600";
const { examples = [], count = 0, reason, protocol, age_days } = finding.params;
const vars = { count, reason: reason ? t(`security.reason.${reason}`) : "", protocol: protocol ?? "", days: age_days ?? "" };
return (
<li className="flex gap-3 px-5 py-3">
<Icon size={18} className={`mt-0.5 shrink-0 ${color}`} aria-hidden />
<div className="min-w-0 flex-1">
<div className="flex flex-wrap items-center gap-2">
<span className="font-medium text-slate-900">{t(`security.checks.${finding.id}.title`)}</span>
{failed && <Badge tone={SEVERITY_TONE[finding.severity]}>{t(`security.severity.${finding.severity}`)}</Badge>}
<span className="sr-only">{finding.passed === null ? t("security.unknown") : failed ? t("security.failed") : t("security.passed")}</span>
</div>
<p className="text-sm text-slate-600">{t(`security.checks.${finding.id}.${failed ? "fail" : finding.passed === null ? "unknown" : "ok"}`, vars)}</p>
{failed && examples.length > 0 && (
<ul className="mt-1 list-disc pl-5 text-sm text-slate-700">
{examples.map((example, i) => <li key={i} className="break-words">{example}</li>)}
{count > examples.length && <li className="list-none text-slate-400">{t("security.andMore", { n: count - examples.length })}</li>}
</ul>
)}
{failed && <p className="mt-1 text-sm text-slate-500">{t(`security.checks.${finding.id}.fix`)}</p>}
</div>
</li>
);
}
+243
View File
@@ -0,0 +1,243 @@
"use client";
import { InfoTip } from "@/components/ui/Tooltip";
import { Badge } from "@/components/ui/Badge";
import { Button } from "@/components/ui/Button";
import { Card } from "@/components/ui/Card";
import { FieldGroup, Input, Select } from "@/components/ui/Field";
import { Modal } from "@/components/ui/Modal";
import { Skeleton } from "@/components/ui/Skeleton";
import { useToast } from "@/components/ui/Toast";
import { api, ApiToken, TokenScope } from "@/lib/api";
import { useLanguage } from "@/components/LanguageProvider";
import { Copy, Plus, Trash2 } from "lucide-react";
import { useEffect, useState } from "react";
const SCOPE_TONE: Record<TokenScope, "neutral" | "info" | "warning"> = { read: "neutral", deploy: "info", admin: "warning" };
export default function TokensPage() {
const { t, errorMessage, locale } = useLanguage();
const toast = useToast();
const [tokens, setTokens] = useState<ApiToken[] | null>(null);
const [isAdmin, setIsAdmin] = useState(false);
const [error, setError] = useState<string | null>(null);
const [showCreate, setShowCreate] = useState(false);
const [created, setCreated] = useState<(ApiToken & { token: string }) | null>(null);
async function load() {
try {
const [list, me] = await Promise.all([api.listApiTokens(), api.me()]);
setTokens(list);
setIsAdmin(me.role === "admin");
setError(null);
} catch (e) {
setError(errorMessage(e, "tokens.loadFailed"));
}
}
useEffect(() => {
load();
}, []);
async function handleRevoke(token: ApiToken) {
if (!confirm(t("tokens.revokeConfirm", { name: token.name }))) return;
try {
await api.revokeApiToken(token.id);
await load();
toast.show(t("tokens.revoked", { name: token.name }), "success");
} catch (e) {
setError(errorMessage(e, "common.deleteFailed"));
}
}
const when = (value: string | null) => (value ? new Date(value).toLocaleString(locale) : null);
return (
<div>
<div className="mb-6 flex items-center justify-between">
<h1 className="flex items-center gap-2 text-2xl font-semibold text-slate-900">
{t("tokens.title")}
<InfoTip text={t("tokens.intro")} size={18} />
</h1>
<Button onClick={() => setShowCreate(true)}>
<Plus size={16} />
{t("tokens.new")}
</Button>
</div>
{error && <p className="mb-4 text-sm text-red-600">{error}</p>}
<Card className="overflow-x-auto">
<table className="w-full text-left text-sm">
<thead className="border-b border-slate-200 bg-slate-50 text-xs uppercase tracking-wide text-slate-500">
<tr>
<th className="px-5 py-3 font-medium">{t("tokens.colName")}</th>
<th className="px-5 py-3 font-medium">{t("tokens.colScope")}</th>
{isAdmin && <th className="px-5 py-3 font-medium">{t("tokens.colOwner")}</th>}
<th className="px-5 py-3 font-medium">{t("tokens.colExpires")}</th>
<th className="px-5 py-3 font-medium">{t("tokens.colLastUsed")}</th>
<th className="px-5 py-3 font-medium"><span className="sr-only">{t("common.actions")}</span></th>
</tr>
</thead>
<tbody>
{tokens === null &&
[0, 1].map((i) => (
<tr key={i} className="border-b border-slate-100 last:border-0">
<td className="px-5 py-3"><Skeleton className="h-4 w-28" /></td>
<td className="px-5 py-3"><Skeleton className="h-4 w-14 rounded-full" /></td>
<td className="px-5 py-3"><Skeleton className="h-4 w-24" /></td>
<td className="px-5 py-3"><Skeleton className="h-4 w-24" /></td>
<td className="px-5 py-3" />
</tr>
))}
{tokens?.length === 0 && (
<tr>
<td colSpan={6} className="px-5 py-8 text-center text-slate-500">{t("tokens.empty")}</td>
</tr>
)}
{tokens?.map((token) => (
<tr key={token.id} className="border-b border-slate-100 last:border-0">
<td className="px-5 py-3">
<div className="font-medium text-slate-900">{token.name}</div>
<div className="font-mono text-xs text-slate-400">{token.prefix}…</div>
</td>
<td className="px-5 py-3"><Badge tone={SCOPE_TONE[token.scope]}>{t(`tokens.scope.${token.scope}`)}</Badge></td>
{isAdmin && <td className="px-5 py-3 text-slate-600">{token.owner}</td>}
<td className="px-5 py-3 text-slate-500">{when(token.expires_at) ?? t("tokens.never")}</td>
<td className="px-5 py-3 text-slate-500">
{when(token.last_used_at) ?? <span className="text-slate-300">{t("tokens.neverUsed")}</span>}
</td>
<td className="px-5 py-3 text-right">
<Button variant="dangerSubtle" onClick={() => handleRevoke(token)}>
<Trash2 size={14} />
{t("tokens.revoke")}
</Button>
</td>
</tr>
))}
</tbody>
</table>
</Card>
{showCreate && (
<CreateTokenModal
isAdmin={isAdmin}
onClose={() => setShowCreate(false)}
onCreated={async (token) => {
setShowCreate(false);
setCreated(token);
await load();
}}
/>
)}
{created && <TokenCreatedModal token={created} onClose={() => setCreated(null)} />}
</div>
);
}
function CreateTokenModal({
isAdmin,
onClose,
onCreated,
}: {
isAdmin: boolean;
onClose: () => void;
onCreated: (token: ApiToken & { token: string }) => void;
}) {
const { t, errorMessage } = useLanguage();
const [name, setName] = useState("");
const [scope, setScope] = useState<TokenScope>("read");
const [days, setDays] = useState("90");
const [password, setPassword] = useState("");
const [saving, setSaving] = useState(false);
const [error, setError] = useState<string | null>(null);
async function handleSubmit(e: React.FormEvent) {
e.preventDefault();
setSaving(true);
setError(null);
try {
const token = await api.createApiToken({
name: name.trim(),
scope,
expires_in_days: days === "never" ? null : Number(days),
current_password: password,
});
onCreated(token);
} catch (err) {
setError(errorMessage(err, "tokens.createFailed"));
} finally {
setSaving(false);
}
}
return (
<Modal title={t("tokens.new")} onClose={onClose}>
<form onSubmit={handleSubmit} className="space-y-4">
<FieldGroup label={t("tokens.colName")} hint={t("tokens.nameHint")}>
<Input required maxLength={80} value={name} onChange={(e) => setName(e.target.value)} />
</FieldGroup>
<FieldGroup label={t("tokens.colScope")} hint={t(`tokens.scopeHint.${scope}`)}>
<Select value={scope} onChange={(e) => setScope(e.target.value as TokenScope)}>
<option value="read">{t("tokens.scope.read")}</option>
{isAdmin && <option value="deploy">{t("tokens.scope.deploy")}</option>}
{isAdmin && <option value="admin">{t("tokens.scope.admin")}</option>}
</Select>
</FieldGroup>
<FieldGroup label={t("tokens.colExpires")}>
<Select value={days} onChange={(e) => setDays(e.target.value)}>
<option value="30">{t("tokens.days", { n: 30 })}</option>
<option value="90">{t("tokens.days", { n: 90 })}</option>
<option value="365">{t("tokens.days", { n: 365 })}</option>
<option value="never">{t("tokens.never")}</option>
</Select>
</FieldGroup>
<FieldGroup label={t("tokens.password")} hint={t("tokens.passwordHint")}>
<Input required type="password" autoComplete="current-password" maxLength={1024} value={password} onChange={(e) => setPassword(e.target.value)} />
</FieldGroup>
{error && <p className="text-sm text-red-600">{error}</p>}
<div className="flex justify-end gap-2 pt-2">
<Button type="button" variant="ghost" onClick={onClose}>
{t("common.cancel")}
</Button>
<Button type="submit" disabled={saving} loading={saving}>
{saving ? t("common.saving") : t("tokens.create")}
</Button>
</div>
</form>
</Modal>
);
}
function TokenCreatedModal({ token, onClose }: { token: ApiToken & { token: string }; onClose: () => void }) {
const { t } = useLanguage();
const toast = useToast();
const example = `curl -H "Authorization: Bearer ${token.token}" ${typeof window === "undefined" ? "" : window.location.origin}/api/hosts`;
function copy(text: string) {
void navigator.clipboard?.writeText(text);
toast.show(t("tokens.copied"), "success");
}
return (
<Modal title={t("tokens.createdTitle", { name: token.name })} onClose={onClose}>
<div className="space-y-4">
<p className="text-sm text-amber-700">{t("tokens.shownOnce")}</p>
<div className="flex items-start gap-2">
<code className="flex-1 break-all rounded-md border border-slate-200 bg-slate-50 p-2 font-mono text-xs text-slate-900">{token.token}</code>
<Button type="button" variant="outline" onClick={() => copy(token.token)}>
<Copy size={14} />
{t("tokens.copy")}
</Button>
</div>
<div>
<p className="mb-1 text-xs font-medium uppercase tracking-wide text-slate-500">{t("tokens.example")}</p>
<code className="block break-all rounded-md border border-slate-200 bg-slate-50 p-2 font-mono text-xs text-slate-700">{example}</code>
</div>
<div className="flex justify-end pt-2">
<Button onClick={onClose}>{t("tokens.done")}</Button>
</div>
</div>
</Modal>
);
}
+8
View File
@@ -29,6 +29,7 @@ import {
Server,
Settings,
Shield,
ShieldAlert,
ShieldCheck,
Tags,
Users,
@@ -121,6 +122,7 @@ export const navGroups: NavGroup[] = [
icon: ScrollText,
items: [
{ href: "/logs", label: "Logs", icon: ScrollText, keywords: ["Firewall-Logs", "Geblockt", "Blocked", "Top Ports"] },
{ href: "/security", label: "Sicherheitslage", icon: ShieldAlert, keywords: ["Compliance", "Härtung", "Score", "Regel-Hygiene", "Any-Any", "SSH", "Sicherheit", "Prüfung"] },
{ href: "/audit", label: "Audit", icon: ScrollText, keywords: ["Deploy-Historie", "Protokoll", "Verlauf"] },
],
},
@@ -163,6 +165,12 @@ export const navGroups: NavGroup[] = [
icon: Users,
keywords: ["User", "Accounts", "Rollen", "Rechte", "Read-Only", "Nur-Lesen", "Admin-Rolle", "Passwort zurücksetzen"],
},
{
href: "/tokens",
label: "API-Tokens",
icon: KeyRound,
keywords: ["API", "Token", "Bearer", "Prometheus", "Metriken", "Automatisierung", "CI", "Skript", "MCP", "Schlüssel"],
},
{
href: "/setup",
label: "Einrichtung",
+46
View File
@@ -670,6 +670,42 @@ export interface AuthState {
role?: UserRole;
}
export type TokenScope = "read" | "deploy" | "admin";
export interface ApiToken {
id: string;
name: string;
scope: TokenScope;
prefix: string;
owner: string;
created_at: string;
expires_at: string | null;
last_used_at: string | null;
last_used_ip: string | null;
}
export type Severity = "critical" | "high" | "medium" | "low" | "info";
export interface ComplianceFinding {
id: string;
severity: Severity;
passed: boolean | null;
params: { count?: number; examples?: string[]; reason?: string; protocol?: string; age_days?: number | null };
}
export interface ComplianceHost {
host_id: string;
host_name: string;
cluster_name: string | null;
enabled: boolean;
checked_at: string | null;
score: number | null;
grade: "good" | "fair" | "poor" | null;
error: "unreachable" | "invalid_config" | null;
counts: Record<Severity, number> | null;
findings?: ComplianceFinding[];
}
export interface User {
id: string;
username: string;
@@ -1286,6 +1322,16 @@ export const api = {
request<User>(`/users/${id}/reset-password`, { method: "POST", body: JSON.stringify({ new_password }) }),
deleteUser: (id: string) => request<void>(`/users/${id}`, { method: "DELETE" }),
listCompliance: () => request<ComplianceHost[]>("/compliance"),
getHostCompliance: (id: string) => request<ComplianceHost>(`/hosts/${id}/compliance`),
checkHostCompliance: (id: string) => request<ComplianceHost>(`/hosts/${id}/compliance/check`, { method: "POST" }),
checkFleetCompliance: () => request<ComplianceHost[]>("/compliance/check", { method: "POST" }),
listApiTokens: () => request<ApiToken[]>("/tokens"),
createApiToken: (data: { name: string; scope: TokenScope; expires_in_days: number | null; current_password: string }) =>
request<ApiToken & { token: string }>("/tokens", { method: "POST", body: JSON.stringify(data) }),
revokeApiToken: (id: string) => request<void>(`/tokens/${id}`, { method: "DELETE" }),
listPolicyHierarchyGroups: () => request<PolicyHierarchyGroup[]>("/policy-hierarchy/groups"),
createPolicyHierarchyGroup: (data: PolicyHierarchyGroupWrite) => request<PolicyHierarchyGroup>("/policy-hierarchy/groups", { method: "POST", body: JSON.stringify(data) }),
updatePolicyHierarchyGroup: (id: string, data: PolicyHierarchyGroupWrite) => request<PolicyHierarchyGroup>(`/policy-hierarchy/groups/${id}`, { method: "PUT", body: JSON.stringify(data) }),
+165 -1
View File
@@ -212,9 +212,11 @@
"certificates": "Zertifikate",
"logs": "Logs",
"audit": "Audit",
"security": "Sicherheitslage",
"updates": "Update Center",
"settings": "Einstellungen",
"users": "Benutzer",
"tokens": "API-Tokens",
"network-devices": "Geräte",
"network-compare": "Abgleich",
"hostsystem-tunables": "Tunables",
@@ -579,7 +581,11 @@
"UPDATER_REJECTED": "Der Updater hat die Anfrage abgelehnt ({reason}).",
"UPDATE_NOT_OFFERED": "Diese Version wird nicht zur Installation angeboten — nur die neueste stabile Version.",
"UPDATE_MAJOR_CHANGE": "Hauptversions-Sprünge bitte bewusst auf dem Server mit „opnexus update“ ausführen.",
"ADMIN_REQUIRED": "Diese Funktion erfordert die Rolle „Admin“."
"ADMIN_REQUIRED": "Diese Funktion erfordert die Rolle „Admin“.",
"API_TOKEN_LIMIT_REACHED": "Es sind höchstens {limit} API-Tokens pro Benutzer möglich. Widerrufe zuerst nicht mehr benötigte Tokens.",
"API_TOKEN_NOT_FOUND": "API-Token nicht gefunden",
"API_TOKEN_SCOPE_FORBIDDEN": "Dieses API-Token (Bereich „{scope}“) darf diese Aktion nicht ausführen.",
"API_TOKEN_SCOPE_TOO_HIGH": "Nur-Lese-Konten können nur Tokens mit dem Bereich „Lesen“ anlegen."
},
"workflow": {
"attention": "Fehler prüfen",
@@ -1424,6 +1430,164 @@
"peerPsk": "Peer-PSK",
"pskPlaceholder": "PSK einfügen"
},
"security": {
"title": "Sicherheitslage",
"intro": "Prüft die Konfiguration jeder Firewall auf typische Schwachstellen (offene Regeln, schwaches SSH/SNMP, fehlende Backups …) und vergibt einen Wert von 0 bis 100. Nur lesend: an den Firewalls wird nichts geändert. Automatisch alle 6 Stunden geprüft.",
"loadFailed": "Sicherheitslage konnte nicht geladen werden",
"checkFailed": "Die Prüfung ist fehlgeschlagen",
"checked": "Prüfung abgeschlossen",
"checkAll": "Alle prüfen",
"checkHost": "Erneut prüfen",
"noHosts": "Noch keine Firewalls",
"noHostsHint": "Verbinde eine Firewall, um ihre Sicherheitslage zu sehen.",
"notChecked": "Nicht geprüft",
"neverChecked": "Noch nicht geprüft",
"checkedAt": "Geprüft {when}",
"allGood": "Keine offenen Befunde",
"andMore": "… und {n} weitere",
"unknown": "unbekannt",
"failed": "nicht bestanden",
"passed": "bestanden",
"severity": {
"critical": "kritisch",
"high": "hoch",
"medium": "mittel",
"low": "niedrig",
"info": "Hinweis"
},
"error": {
"unreachable": "nicht erreichbar",
"invalid_config": "nicht lesbar"
},
"errorHint": {
"unreachable": "Die Firewall war nicht erreichbar und konnte deshalb nicht geprüft werden.",
"invalid_config": "Die Konfiguration konnte nicht gelesen werden."
},
"reason": {
"root_password": "root darf sich per Passwort über SSH anmelden",
"password": "SSH erlaubt die Anmeldung per Passwort"
},
"checks": {
"webgui_https": {
"title": "Web-Oberfläche nur über HTTPS",
"ok": "Die Verwaltungsoberfläche läuft über HTTPS.",
"fail": "Die Verwaltungsoberfläche läuft über unverschlüsseltes HTTP. Zugangsdaten sind im Netz mitlesbar.",
"fix": "System › Einstellungen › Administration: Protokoll auf HTTPS stellen."
},
"ssh_hardened": {
"title": "SSH abgesichert",
"ok": "SSH ist aus oder erlaubt nur Schlüssel.",
"fail": "{reason}.",
"fix": "SSH ausschalten, oder unter System › Einstellungen › Administration Root-Login und Passwort-Anmeldung abschalten und Schlüssel nutzen."
},
"snmp_community": {
"title": "SNMP-Community ist nicht „public“",
"ok": "SNMP nutzt eine eigene Community.",
"fail": "SNMP nutzt die voreingestellte Community „public“. Jeder im Netz kann Systemdaten lesen.",
"fix": "Eine eigene, lange Community setzen oder SNMP ausschalten."
},
"wan_any_any": {
"title": "Keine Allow-all-Regel am WAN",
"ok": "Am WAN erlaubt keine Regel allen Verkehr überallhin.",
"fail": "{count} Regel(n) am WAN erlauben jeden Verkehr zu jedem Ziel — die Firewall schützt dort nichts.",
"fix": "Regel auf die nötigen Quellen, Ziele und Ports beschränken oder löschen."
},
"internal_any_any": {
"title": "Keine Allow-all-Regel mit beliebiger Quelle (intern)",
"ok": "Keine interne Regel erlaubt beliebige Quellen zu beliebigen Zielen.",
"fail": "{count} Regel(n) erlauben beliebige Quellen zu beliebigen Zielen.",
"fix": "Die Quelle auf das jeweilige Netz beschränken."
},
"rules_shadowed": {
"title": "Keine verdeckten Regeln",
"ok": "Jede Regel wird erreicht.",
"fail": "{count} Regel(n) werden nie erreicht, weil eine frühere Regel sie schon abdeckt.",
"fix": "Reihenfolge prüfen; die überflüssige Regel entfernen oder davor einsortieren."
},
"rules_duplicate": {
"title": "Keine doppelten Regeln",
"ok": "Keine Regel kommt doppelt vor.",
"fail": "{count} Regel(n) sind identisch zu einer früheren.",
"fix": "Die doppelte Regel löschen."
},
"rules_documented": {
"title": "Regeln sind beschriftet",
"ok": "Alle aktiven Regeln haben eine Beschreibung.",
"fail": "{count} aktive Regel(n) haben keine Beschreibung.",
"fix": "Beschreibung ergänzen, damit später klar bleibt, wofür die Regel da ist."
},
"ntp_configured": {
"title": "Zeitserver gesetzt",
"ok": "Zeitserver sind eingetragen.",
"fail": "Kein Zeitserver konfiguriert. Logs und Zertifikate brauchen die richtige Uhrzeit.",
"fix": "Zeitserver in den Zeiteinstellungen eintragen."
},
"remote_logging": {
"title": "Logs liegen auch woanders",
"ok": "Ein externes Syslog-Ziel ist eingetragen.",
"fail": "Logs liegen nur auf der Firewall. Nach einem Einbruch oder Ausfall fehlen sie.",
"fix": "Unter System › Einstellungen › Logging einen Syslog-Server eintragen."
},
"ids_enabled": {
"title": "Einbruchserkennung aktiv",
"ok": "Die Einbruchserkennung ist an.",
"fail": "Die Einbruchserkennung (IDS/IPS) ist nicht eingeschaltet.",
"fix": "Optional: Dienste › Intrusion Detection aktivieren."
},
"firmware_current": {
"title": "Firmware aktuell",
"ok": "Keine Updates offen.",
"fail": "Es stehen Updates bereit.",
"fix": "Im Update Center einspielen.",
"unknown": "Der Updatestatus konnte nicht geprüft werden."
},
"backup_recent": {
"title": "Aktuelles Backup",
"ok": "Das neueste Backup ist {days} Tag(e) alt.",
"fail": "Es gibt kein Backup aus den letzten 7 Tagen.",
"fix": "Auf der Host-Seite „Jetzt sichern“ ausführen oder eine Wartung planen."
}
}
},
"tokens": {
"title": "API-Tokens",
"new": "Neues Token",
"intro": "Mit Tokens nutzen Skripte, CI-Jobs, Prometheus und KI-Assistenten die OPNexus-API ohne Anmeldung. Jedes Token gehört dir, hat einen Bereich und lässt sich jederzeit widerrufen. Es bekommt nie mehr Rechte als dein Konto.",
"loadFailed": "API-Tokens konnten nicht geladen werden",
"createFailed": "Token konnte nicht erstellt werden",
"empty": "Noch keine Tokens.",
"colName": "Name",
"colScope": "Bereich",
"colOwner": "Besitzer",
"colExpires": "Läuft ab",
"colLastUsed": "Zuletzt benutzt",
"never": "Nie",
"neverUsed": "Noch nicht benutzt",
"days": "{n} Tage",
"nameHint": "Wofür das Token gedacht ist, z. B. „Grafana“ oder „CI-Deploy“.",
"scope": {
"read": "Lesen",
"deploy": "Ausrollen",
"admin": "Admin"
},
"scopeHint": {
"read": "Nur lesen: alles ansehen, nichts ändern. Für Dashboards, Prometheus und KI-Assistenten.",
"deploy": "Lesen plus Regeln, Aliase, NAT, VPN und Wartung ausrollen. Kann keine Benutzer, Einstellungen oder Hosts verwalten.",
"admin": "Alles, was dein Konto darf, außer Tokens und dein eigenes Konto zu verwalten."
},
"password": "Dein Passwort",
"passwordHint": "Zum Erstellen eines Tokens ist dein aktuelles Passwort nötig.",
"create": "Token erstellen",
"revoke": "Widerrufen",
"revokeConfirm": "Token „{name}“ widerrufen? Alles, was es nutzt, funktioniert sofort nicht mehr.",
"revoked": "Token „{name}“ widerrufen",
"createdTitle": "Token „{name}“ erstellt",
"shownOnce": "Kopiere das Token jetzt. Es wird nur einmal angezeigt und lässt sich nicht erneut anzeigen.",
"copy": "Kopieren",
"copied": "Kopiert",
"example": "Beispiel",
"done": "Fertig"
},
"users": {
"roleAdmin": "Admin",
"roleReadOnly": "Nur-Lesen",
+165 -1
View File
@@ -212,9 +212,11 @@
"certificates": "Certificates",
"logs": "Logs",
"audit": "Audit",
"security": "Security posture",
"updates": "Update Center",
"settings": "Settings",
"users": "Users",
"tokens": "API tokens",
"network-devices": "Devices",
"network-compare": "Compare",
"hostsystem-tunables": "Tunables",
@@ -579,7 +581,11 @@
"UPDATER_REJECTED": "The updater rejected the request ({reason}).",
"UPDATE_NOT_OFFERED": "This version is not offered for installation — only the latest stable version.",
"UPDATE_MAJOR_CHANGE": "Please run major version jumps deliberately on the server with “opnexus update”.",
"ADMIN_REQUIRED": "This function requires the “Admin” role."
"ADMIN_REQUIRED": "This function requires the “Admin” role.",
"API_TOKEN_LIMIT_REACHED": "At most {limit} API tokens are allowed per user. Revoke tokens you no longer need first.",
"API_TOKEN_NOT_FOUND": "API token not found",
"API_TOKEN_SCOPE_FORBIDDEN": "This API token (scope “{scope}”) is not allowed to perform this action.",
"API_TOKEN_SCOPE_TOO_HIGH": "Read-only accounts can only create tokens with the “read” scope."
},
"workflow": {
"attention": "Check errors",
@@ -1424,6 +1430,164 @@
"peerPsk": "Peer PSK",
"pskPlaceholder": "Paste PSK"
},
"security": {
"title": "Security posture",
"intro": "Checks every firewall's configuration for common weaknesses (open rules, weak SSH/SNMP, missing backups…) and gives a score from 0 to 100. Read-only: nothing is changed on the firewalls. Checked automatically every 6 hours.",
"loadFailed": "Could not load the security posture",
"checkFailed": "The check failed",
"checked": "Check finished",
"checkAll": "Check all",
"checkHost": "Check again",
"noHosts": "No firewalls yet",
"noHostsHint": "Connect a firewall to see its security posture.",
"notChecked": "Not checked",
"neverChecked": "Not checked yet",
"checkedAt": "Checked {when}",
"allGood": "No open findings",
"andMore": "… and {n} more",
"unknown": "unknown",
"failed": "failed",
"passed": "passed",
"severity": {
"critical": "critical",
"high": "high",
"medium": "medium",
"low": "low",
"info": "note"
},
"error": {
"unreachable": "unreachable",
"invalid_config": "unreadable"
},
"errorHint": {
"unreachable": "The firewall could not be reached, so it could not be checked.",
"invalid_config": "The configuration could not be read."
},
"reason": {
"root_password": "root may log in over SSH with a password",
"password": "SSH accepts password login"
},
"checks": {
"webgui_https": {
"title": "Web interface over HTTPS only",
"ok": "The management interface is served over HTTPS.",
"fail": "The management interface runs over unencrypted HTTP. Credentials can be read on the network.",
"fix": "System › Settings › Administration: set the protocol to HTTPS."
},
"ssh_hardened": {
"title": "SSH locked down",
"ok": "SSH is off or accepts keys only.",
"fail": "{reason}.",
"fix": "Switch SSH off, or disable root login and password login under System › Settings › Administration and use keys."
},
"snmp_community": {
"title": "SNMP community is not “public”",
"ok": "SNMP uses a custom community.",
"fail": "SNMP uses the default community “public”. Anyone on the network can read system data.",
"fix": "Set your own long community or turn SNMP off."
},
"wan_any_any": {
"title": "No allow-all rule on the WAN",
"ok": "No rule on the WAN allows all traffic to everywhere.",
"fail": "{count} rule(s) on the WAN allow all traffic to any destination — the firewall protects nothing there.",
"fix": "Restrict the rule to the sources, destinations and ports you need, or delete it."
},
"internal_any_any": {
"title": "No allow-all rule from any source (internal)",
"ok": "No internal rule allows any source to any destination.",
"fail": "{count} rule(s) allow any source to any destination.",
"fix": "Limit the source to the network the rule is meant for."
},
"rules_shadowed": {
"title": "No shadowed rules",
"ok": "Every rule can be reached.",
"fail": "{count} rule(s) are never reached because an earlier rule already covers them.",
"fix": "Check the order; remove the useless rule or move it up."
},
"rules_duplicate": {
"title": "No duplicate rules",
"ok": "No rule is repeated.",
"fail": "{count} rule(s) are identical to an earlier one.",
"fix": "Delete the duplicate."
},
"rules_documented": {
"title": "Rules are described",
"ok": "All active rules have a description.",
"fail": "{count} active rule(s) have no description.",
"fix": "Add a description so it stays clear later what the rule is for."
},
"ntp_configured": {
"title": "Time server set",
"ok": "Time servers are configured.",
"fail": "No time server is configured. Logs and certificates need the right time.",
"fix": "Enter NTP servers in the time settings."
},
"remote_logging": {
"title": "Logs are kept elsewhere too",
"ok": "A remote syslog target is configured.",
"fail": "Logs exist only on the firewall. After a break-in or a failure they are gone.",
"fix": "Add a syslog server under System › Settings › Logging."
},
"ids_enabled": {
"title": "Intrusion detection active",
"ok": "Intrusion detection is on.",
"fail": "Intrusion detection (IDS/IPS) is not switched on.",
"fix": "Optional: enable Services › Intrusion Detection."
},
"firmware_current": {
"title": "Firmware up to date",
"ok": "No updates are pending.",
"fail": "Updates are available.",
"fix": "Install them in the Update Center.",
"unknown": "Could not determine the update status."
},
"backup_recent": {
"title": "Recent backup",
"ok": "The newest backup is {days} day(s) old.",
"fail": "There is no backup from the last 7 days.",
"fix": "Use “Back up now” on the host page or schedule maintenance."
}
}
},
"tokens": {
"title": "API tokens",
"new": "New token",
"intro": "Tokens let scripts, CI jobs, Prometheus and AI assistants use the OPNexus API without a sign-in. Each token belongs to you, has a scope and can be revoked at any time. It never gets more rights than your account.",
"loadFailed": "Could not load API tokens",
"createFailed": "Could not create the token",
"empty": "No tokens yet.",
"colName": "Name",
"colScope": "Scope",
"colOwner": "Owner",
"colExpires": "Expires",
"colLastUsed": "Last used",
"never": "Never",
"neverUsed": "Not used yet",
"days": "{n} days",
"nameHint": "What the token is for, e.g. “Grafana” or “CI deploy”.",
"scope": {
"read": "Read",
"deploy": "Deploy",
"admin": "Admin"
},
"scopeHint": {
"read": "Read only: view everything, change nothing. For dashboards, Prometheus and AI assistants.",
"deploy": "Read plus roll out rules, aliases, NAT, VPN and maintenance. Cannot manage users, settings or hosts.",
"admin": "Everything your account may do, except managing tokens and your own account."
},
"password": "Your password",
"passwordHint": "Creating a token needs your current password.",
"create": "Create token",
"revoke": "Revoke",
"revokeConfirm": "Revoke token “{name}”? Anything using it stops working immediately.",
"revoked": "Token “{name}” revoked",
"createdTitle": "Token “{name}” created",
"shownOnce": "Copy the token now. It is shown only once and cannot be displayed again.",
"copy": "Copy",
"copied": "Copied",
"example": "Example",
"done": "Done"
},
"users": {
"roleAdmin": "Admin",
"roleReadOnly": "Read-only",
+2 -2
View File
@@ -1,6 +1,6 @@
// Manually bumped whenever a notable change ships — mirrors the dated
// entries already kept in the project README/CHANGELOG.md, no build
// tooling involved. Versioning scheme: see CHANGELOG.md ("Versionierung").
export const APP_VERSION = "1.17.0";
export const APP_CODENAME = "Imperial";
export const APP_VERSION = "1.18.0";
export const APP_CODENAME = "Kestrel";
export const APP_MAINTAINER = "Julian Boldt";