diff --git a/.env.example b/.env.example index 2f12ec3..a91aa2e 100644 --- a/.env.example +++ b/.env.example @@ -37,9 +37,21 @@ NETX_UME_ALARM_WS_ENABLED=true NETX_UME_NOTIFICATION_ESTABLISH_PATH=/restconf/operations/zte-notifications:establish-subscription NETX_UME_NOTIFICATION_DELETE_PATH=/restconf/operations/zte-notifications:delete-subscription NETX_UME_NOTIFICATION_TOPIC=ALARM -# Auth (optional — lab defaults: admin/admin123 + data/auth/mcp_token) +# TLS verify for UME (default true). Set false only for lab self-signed certs. +# NETX_UME_VERIFY_TLS=true +# Auth (lab defaults: admin/admin123 + data/auth/mcp_token) # NETX_AUTH_ENABLED=true -# NETX_AUTH_SECRET=change-me-in-production +# Leave NETX_AUTH_SECRET empty to auto-create data/auth/jwt_secret on first boot. +# NETX_AUTH_SECRET= +# NETX_AUTH_SECRET_FILE=data/auth/jwt_secret # NETX_BOOTSTRAP_ADMIN_USERNAME=admin # NETX_BOOTSTRAP_ADMIN_PASSWORD=admin123 # NETX_API_TOKEN= # MCP: leave empty to auto-read data/auth/mcp_token +# API docs (/docs, /redoc, /openapi.json); default false. Lab: set true. +# NETX_DOCS_ENABLED=true +# NETX_ALLOW_INSECURE_DEFAULTS=false +# NETX_SKIP_LEGACY_STARTUP_DDL=true +# NETX_SQL_READONLY_DATABASE_URL=postgresql+psycopg://netx_ro:xxx@127.0.0.1:5432/netx +# NETX_RUN_INLINE_SCHEDULERS=true +# NETX_AUDIT_ASYNC=true +# NETX_AUDIT_SAMPLE_N=1 diff --git a/PROD_MIN_CHECKLIST.md b/PROD_MIN_CHECKLIST.md index 9a23004..fc2584f 100644 --- a/PROD_MIN_CHECKLIST.md +++ b/PROD_MIN_CHECKLIST.md @@ -1,19 +1,27 @@ # netx Production Minimum Checklist ## Security -- Replace local shared token with a strong secret and rotate it regularly. +- Prefer empty `NETX_AUTH_SECRET` so each install auto-writes `data/auth/jwt_secret` (do not commit that file). Set an explicit secret only for multi-node shared signing. +- Leave `NETX_DOCS_ENABLED` unset/false so `/docs` and OpenAPI stay off (set `true` only in lab). +- Keep `NETX_UME_VERIFY_TLS=true` (or pin a CA); avoid `false` on non-lab hosts. +- Binding `NETX_HOST` to a non-loopback address with lab defaults is refused unless `NETX_ALLOW_INSECURE_DEFAULTS=1`. +- Prefer scoped API tokens (MCP default excludes `webcrt:session` / `sql:query`). - Keep `.env` and `oclaw/_local/system.env` out of Git (already ignored). - Restrict access to `127.0.0.1` or internal network only. ## Runtime - Ensure PostgreSQL backup policy exists (daily logical backup + retention). +- Prefer Alembic: `alembic upgrade head` and `NETX_SKIP_LEGACY_STARTUP_DDL=true`. +- Optional: `NETX_RUN_INLINE_SCHEDULERS=false` and run `python -m netx_api.worker` for collectors. - Run `oclaw` and `netx` under process managers (systemd/Windows service/pm2 equivalent). - Enable auto-restart and startup-at-boot for both services. ## Observability - Health checks: - `oclaw`: `/admin/api/ops-ai/health` (with Bearer token) - - `netx`: `/v1/integrations/status` + - `netx` liveness: `/health/live` + - `netx` readiness: `/health/ready` + - `netx` integrations: `/v1/integrations/status` - Alert when `oclaw_bridge.status != up` for more than 2 polling cycles. - Alert when `db.status != up` or `latency_ms` exceeds threshold. @@ -21,8 +29,7 @@ - Keep one documented restart order: 1) PostgreSQL 2) `oclaw` - 3) `netx` + 3) `netx` (and worker if split) - Validate after restart: - `GET /v1/integrations/status` returns all major components as `up`. - Keep parser config and importer changes versioned and reviewed before release. - diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..222c396 --- /dev/null +++ b/alembic.ini @@ -0,0 +1,42 @@ +# Alembic config for netx + +[alembic] +script_location = alembic +prepend_sys_path = . +version_path_separator = os + +sqlalchemy.url = driver://user:pass@localhost/dbname + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARN +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARN +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/alembic/env.py b/alembic/env.py new file mode 100644 index 0000000..b7bc52c --- /dev/null +++ b/alembic/env.py @@ -0,0 +1,50 @@ +"""Alembic environment — uses NETX_DATABASE_URL / settings.database_url.""" + +from __future__ import annotations + +from logging.config import fileConfig + +from alembic import context +from sqlalchemy import engine_from_config, pool + +from netx_api.config import settings +from netx_api.db import Base +import netx_api.models # noqa: F401 — register metadata + +config = context.config +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +target_metadata = Base.metadata +config.set_main_option("sqlalchemy.url", settings.database_url) + + +def run_migrations_offline() -> None: + url = config.get_main_option("sqlalchemy.url") + context.configure( + url=url, + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + compare_type=True, + ) + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online() -> None: + connectable = engine_from_config( + config.get_section(config.config_ini_section, {}), + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + with connectable.connect() as connection: + context.configure(connection=connection, target_metadata=target_metadata, compare_type=True) + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/alembic/script.py.mako b/alembic/script.py.mako new file mode 100644 index 0000000..d5bc78b --- /dev/null +++ b/alembic/script.py.mako @@ -0,0 +1,28 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} + +""" +from __future__ import annotations + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +# revision identifiers, used by Alembic. +revision: str = ${repr(up_revision)} +down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)} +branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)} +depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} diff --git a/alembic/versions/20260802_scopes.py b/alembic/versions/20260802_scopes.py new file mode 100644 index 0000000..387a88e --- /dev/null +++ b/alembic/versions/20260802_scopes.py @@ -0,0 +1,54 @@ +"""Add app_user.scopes and api_token.scopes for capability RBAC. + +Revision ID: 20260802_scopes +Revises: +Create Date: 2026-08-02 +""" + +from __future__ import annotations + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "20260802_scopes" +down_revision: Union[str, Sequence[str], None] = None +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + bind = op.get_bind() + dialect = bind.dialect.name + # Idempotent adds for brownfield DBs that already ran startup ALTER TABLE. + if dialect == "postgresql": + op.execute("ALTER TABLE app_user ADD COLUMN IF NOT EXISTS scopes JSON DEFAULT '[]'") + op.execute("ALTER TABLE api_token ADD COLUMN IF NOT EXISTS scopes JSON DEFAULT '[]'") + else: + insp = sa.inspect(bind) + if "app_user" in insp.get_table_names(): + cols = {c["name"] for c in insp.get_columns("app_user")} + if "scopes" not in cols: + op.add_column("app_user", sa.Column("scopes", sa.JSON(), nullable=True)) + if "api_token" in insp.get_table_names(): + cols = {c["name"] for c in insp.get_columns("api_token")} + if "scopes" not in cols: + op.add_column("api_token", sa.Column("scopes", sa.JSON(), nullable=True)) + + +def downgrade() -> None: + bind = op.get_bind() + dialect = bind.dialect.name + if dialect == "postgresql": + op.execute("ALTER TABLE api_token DROP COLUMN IF EXISTS scopes") + op.execute("ALTER TABLE app_user DROP COLUMN IF EXISTS scopes") + else: + try: + op.drop_column("api_token", "scopes") + except Exception: + pass + try: + op.drop_column("app_user", "scopes") + except Exception: + pass diff --git a/netx_api/audit_async.py b/netx_api/audit_async.py new file mode 100644 index 0000000..a15abe5 --- /dev/null +++ b/netx_api/audit_async.py @@ -0,0 +1,107 @@ +"""Async / sampled audit writes to reduce DB pressure on every request.""" + +from __future__ import annotations + +import logging +import queue +import threading +from typing import Any + +from .config import settings +from .db import SessionLocal + +_log = logging.getLogger("netx.audit.async") + +_q: queue.SimpleQueue[dict[str, Any]] | None = None +_worker: threading.Thread | None = None +_lock = threading.Lock() +_counter = 0 + + +def _sample_ok() -> bool: + """When sample_n > 1, keep 1/N of http.* audits; always keep auth/security actions.""" + global _counter + n = int(getattr(settings, "audit_sample_n", 1) or 1) + if n <= 1: + return True + _counter += 1 + return (_counter % n) == 0 + + +def _worker_loop() -> None: + assert _q is not None + while True: + item = _q.get() + if item is None: + break + try: + from .auth_service import write_audit + + db = SessionLocal() + try: + write_audit(db, **item) + finally: + db.close() + except Exception: + _log.exception("async audit write failed") + + +def _ensure_worker() -> queue.SimpleQueue: + global _q, _worker + with _lock: + if _q is None: + _q = queue.SimpleQueue() + _worker = threading.Thread(target=_worker_loop, name="netx-audit-writer", daemon=True) + _worker.start() + return _q + + +def enqueue_audit( + *, + action: str, + actor_user_id: str = "", + actor_username: str = "", + method: str = "", + path: str = "", + status_code: int = 0, + client_ip: str = "", + user_agent: str = "", + detail: dict[str, Any] | None = None, +) -> None: + act = str(action or "") + # Always persist auth / security-relevant events. + if act.startswith("auth.") or act.startswith("users.") or act.startswith("api_tokens.") or act.startswith("webcrt."): + pass + elif act.startswith("http.") and not _sample_ok(): + return + payload = { + "action": act, + "actor_user_id": actor_user_id, + "actor_username": actor_username, + "method": method, + "path": path, + "status_code": status_code, + "client_ip": client_ip, + "user_agent": user_agent, + "detail": detail or {}, + } + if not bool(getattr(settings, "audit_async", True)): + from .auth_service import write_audit + + db = SessionLocal() + try: + write_audit(db, **payload) + finally: + db.close() + return + try: + _ensure_worker().put(payload) + except Exception: + _log.exception("audit enqueue failed; falling back to sync") + from .auth_service import write_audit + + db = SessionLocal() + try: + write_audit(db, **payload) + finally: + db.close() diff --git a/netx_api/auth_deps.py b/netx_api/auth_deps.py index 5771878..bf2c910 100644 --- a/netx_api/auth_deps.py +++ b/netx_api/auth_deps.py @@ -1,14 +1,22 @@ -"""FastAPI dependencies for authenticated / admin-only routes.""" +"""FastAPI dependencies for authenticated / admin-only / scope-gated routes.""" from __future__ import annotations -from dataclasses import dataclass -from typing import Annotated +from dataclasses import dataclass, field +from typing import Annotated, Callable from fastapi import Depends, HTTPException, Request from sqlalchemy.orm import Session -from .auth_service import get_user_by_id, resolve_api_token_user +from .auth_scopes import ( + ALL_SCOPES, + effective_token_scopes, + effective_user_scopes, + has_all_scopes, + has_scope, + normalize_scopes, +) +from .auth_service import get_user_by_id, resolve_api_token_row from .auth_tokens import decode_access_token from .config import settings from .db import get_db @@ -19,26 +27,40 @@ from .models import AppUser class AuthContext: user: AppUser auth_via: str # jwt | api_token | disabled + scopes: frozenset[str] = field(default_factory=frozenset) + api_token_id: str = "" def _extract_bearer(request: Request) -> str: auth = str(request.headers.get("authorization") or "").strip() if auth.lower().startswith("bearer "): return auth[7:].strip() - # WebCRT / tools may pass access_token query + # Prefer Header; query access_token is deprecated (WebSocket may still use short-lived tickets). q = request.query_params.get("access_token") return str(q or "").strip() -def resolve_user_from_token(db: Session, token: str) -> tuple[AppUser, str] | None: +def user_scopes(user: AppUser) -> frozenset[str]: + return effective_user_scopes(role=str(user.role or "user"), override=getattr(user, "scopes", None) or []) + + +def resolve_user_from_token(db: Session, token: str) -> tuple[AppUser, str, frozenset[str], str] | None: + """Return (user, via, scopes, api_token_id) or None.""" raw = str(token or "").strip() if not raw: return None if raw.startswith("nxt_"): - user = resolve_api_token_user(db, raw) - if user is None: + row = resolve_api_token_row(db, raw) + if row is None: return None - return user, "api_token" + user = get_user_by_id(db, row.user_id) + if user is None or not user.is_active: + return None + scopes = effective_token_scopes( + user_scopes=user_scopes(user), + token_scopes=getattr(row, "scopes", None) or [], + ) + return user, "api_token", scopes, str(row.id) try: payload = decode_access_token(raw) except Exception: @@ -48,7 +70,7 @@ def resolve_user_from_token(db: Session, token: str) -> tuple[AppUser, str] | No user = get_user_by_id(db, str(payload.get("sub") or "")) if user is None or not user.is_active: return None - return user, "jwt" + return user, "jwt", user_scopes(user), "" def get_optional_user( @@ -57,21 +79,26 @@ def get_optional_user( ) -> AuthContext | None: if not bool(settings.auth_enabled): return None + cached = getattr(request.state, "auth_user", None) + if isinstance(cached, AppUser): + via = str(getattr(request.state, "auth_via", "") or "jwt") + scopes = getattr(request.state, "auth_scopes", None) + if not isinstance(scopes, frozenset): + scopes = user_scopes(cached) + token_id = str(getattr(request.state, "auth_api_token_id", "") or "") + return AuthContext(user=cached, auth_via=via, scopes=scopes, api_token_id=token_id) token = _extract_bearer(request) if not token: - # Middleware may have already attached user - cached = getattr(request.state, "auth_user", None) - if isinstance(cached, AppUser): - via = str(getattr(request.state, "auth_via", "") or "jwt") - return AuthContext(user=cached, auth_via=via) return None resolved = resolve_user_from_token(db, token) if resolved is None: return None - user, via = resolved + user, via, scopes, token_id = resolved request.state.auth_user = user request.state.auth_via = via - return AuthContext(user=user, auth_via=via) + request.state.auth_scopes = scopes + request.state.auth_api_token_id = token_id + return AuthContext(user=user, auth_via=via, scopes=scopes, api_token_id=token_id) def require_user( @@ -79,7 +106,6 @@ def require_user( db: Session = Depends(get_db), ) -> AuthContext: if not bool(settings.auth_enabled): - # Auth disabled: synthesize a system principal for Depends callers. fake = AppUser( id="system", username="system", @@ -88,7 +114,7 @@ def require_user( is_active=True, created_by="auth_disabled", ) - return AuthContext(user=fake, auth_via="disabled") + return AuthContext(user=fake, auth_via="disabled", scopes=ALL_SCOPES) ctx = get_optional_user(request, db) if ctx is None: raise HTTPException(status_code=401, detail="unauthorized") @@ -96,6 +122,20 @@ def require_user( def require_admin(ctx: Annotated[AuthContext, Depends(require_user)]) -> AuthContext: - if ctx.user.role != "admin": + if ctx.user.role != "admin" and not has_scope(ctx.scopes, "admin:users"): raise HTTPException(status_code=403, detail="admin_required") return ctx + + +def require_scopes(*needed: str) -> Callable[..., AuthContext]: + required = normalize_scopes(needed) + + def _dep(ctx: Annotated[AuthContext, Depends(require_user)]) -> AuthContext: + if not has_all_scopes(ctx.scopes, required): + raise HTTPException( + status_code=403, + detail={"error": "insufficient_scope", "required": required, "granted": sorted(ctx.scopes)}, + ) + return ctx + + return _dep diff --git a/netx_api/auth_middleware.py b/netx_api/auth_middleware.py index f8f8f3f..b8876a5 100644 --- a/netx_api/auth_middleware.py +++ b/netx_api/auth_middleware.py @@ -11,6 +11,7 @@ from starlette.requests import Request from starlette.responses import JSONResponse, Response from .auth_deps import resolve_user_from_token +from .auth_scopes import has_scope, required_scope_for_request from .auth_service import write_audit from .config import settings from .db import SessionLocal @@ -21,25 +22,31 @@ _PUBLIC_EXACT = frozenset( { "/", "/health", - "/openapi.json", - "/docs", - "/docs/oauth2-redirect", - "/redoc", + "/health/live", + "/health/ready", "/favicon.ico", "/v1/auth/login", } ) _PUBLIC_PREFIXES = ( - "/docs", - "/redoc", "/assets", ) +def _docs_public() -> bool: + return bool(getattr(settings, "docs_enabled", False)) + + def _is_public(path: str) -> bool: p = str(path or "") if p in _PUBLIC_EXACT: return True + if _docs_public() and ( + p in ("/openapi.json", "/docs", "/docs/oauth2-redirect", "/redoc") + or p.startswith("/docs") + or p.startswith("/redoc") + ): + return True return any(p.startswith(pref) for pref in _PUBLIC_PREFIXES) @@ -82,7 +89,9 @@ class AuthAuditMiddleware(BaseHTTPMiddleware): auth = str(request.headers.get("authorization") or "").strip() if auth.lower().startswith("bearer "): token = auth[7:].strip() - if not token: + # Query access_token: only allow for non-webcrt paths as deprecated fallback; + # WebCRT HTTP must use Authorization header (see webcrt_router). + if not token and not path.startswith("/v1/webcrt"): token = str(request.query_params.get("access_token") or "").strip() db = SessionLocal() @@ -100,9 +109,35 @@ class AuthAuditMiddleware(BaseHTTPMiddleware): detail={}, ) return JSONResponse(status_code=401, content={"detail": "unauthorized"}) - user, via = resolved + user, via, scopes, token_id = resolved + need = required_scope_for_request(request.method, path) + if need and not has_scope(scopes, need): + write_audit( + db, + action="auth.forbidden_scope", + actor_user_id=str(user.id), + actor_username=str(user.username), + method=request.method, + path=path, + status_code=403, + client_ip=_client_ip(request), + user_agent=str(request.headers.get("user-agent") or "")[:512], + detail={"required": need, "granted": sorted(scopes), "auth_via": via}, + ) + return JSONResponse( + status_code=403, + content={ + "detail": { + "error": "insufficient_scope", + "required": [need], + "granted": sorted(scopes), + } + }, + ) request.state.auth_user = user request.state.auth_via = via + request.state.auth_scopes = scopes + request.state.auth_api_token_id = token_id actor_id = str(user.id) actor_name = str(user.username) auth_via = via @@ -115,25 +150,22 @@ class AuthAuditMiddleware(BaseHTTPMiddleware): started = time.perf_counter() response = await call_next(request) try: - db2 = SessionLocal() - try: - write_audit( - db2, - action=_action_for(request.method, path), - actor_user_id=actor_id, - actor_username=actor_name, - method=request.method, - path=path, - status_code=int(response.status_code), - client_ip=_client_ip(request), - user_agent=str(request.headers.get("user-agent") or "")[:512], - detail={ - "auth_via": auth_via, - "elapsed_ms": int((time.perf_counter() - started) * 1000), - }, - ) - finally: - db2.close() + from .audit_async import enqueue_audit + + enqueue_audit( + action=_action_for(request.method, path), + actor_user_id=actor_id, + actor_username=actor_name, + method=request.method, + path=path, + status_code=int(response.status_code), + client_ip=_client_ip(request), + user_agent=str(request.headers.get("user-agent") or "")[:512], + detail={ + "auth_via": auth_via, + "elapsed_ms": int((time.perf_counter() - started) * 1000), + }, + ) except Exception: - _log.exception("audit write after request failed path=%s", path) + _log.exception("audit enqueue after request failed path=%s", path) return response diff --git a/netx_api/auth_router.py b/netx_api/auth_router.py index f465bc0..c15d38b 100644 --- a/netx_api/auth_router.py +++ b/netx_api/auth_router.py @@ -99,7 +99,14 @@ def api_logout( @router.get("/v1/auth/me") def api_me(ctx: Annotated[AuthContext, Depends(require_user)]) -> dict[str, Any]: - return {"user": user_public(ctx.user), "auth_via": ctx.auth_via} + from .auth_scopes import ALL_SCOPES + + return { + "user": user_public(ctx.user), + "auth_via": ctx.auth_via, + "scopes": sorted(ctx.scopes), + "all_scopes": sorted(ALL_SCOPES), + } @router.post("/v1/auth/change-password") @@ -145,6 +152,7 @@ def api_create_user( password=body.password, role=body.role, actor=ctx.user, + scopes=body.scopes, ) ip, ua = _client_meta(request) write_audit( @@ -177,6 +185,7 @@ def api_update_user( is_active=body.is_active, role=body.role, password=body.password, + scopes=body.scopes, ) ip, ua = _client_meta(request) write_audit( @@ -258,6 +267,7 @@ def api_create_token( user=target, name=body.name, expires_in_days=expires_in_days, + scopes=body.scopes, ) ip, ua = _client_meta(request) write_audit( @@ -275,6 +285,7 @@ def api_create_token( "name": row.name, "owner_user_id": target.id, "owner_username": target.username, + "scopes": getattr(row, "scopes", None) or [], "expires_at": row.expires_at.isoformat() if row.expires_at else None, }, ) @@ -284,6 +295,7 @@ def api_create_token( "name": row.name, "user_id": row.user_id, "username": target.username, + "scopes": getattr(row, "scopes", None) or [], "created_at": row.created_at.isoformat() if row.created_at else None, "expires_at": row.expires_at.isoformat() if row.expires_at else None, "token": plaintext, diff --git a/netx_api/auth_schemas.py b/netx_api/auth_schemas.py index 77da0b9..404673e 100644 --- a/netx_api/auth_schemas.py +++ b/netx_api/auth_schemas.py @@ -19,12 +19,14 @@ class UserCreateRequest(BaseModel): username: str = Field(min_length=2, max_length=64) password: str = Field(min_length=6, max_length=256) role: str = Field(default="user") + scopes: list[str] | None = None class UserUpdateRequest(BaseModel): is_active: bool | None = None role: str | None = None password: str | None = Field(default=None, min_length=6, max_length=256) + scopes: list[str] | None = None class ApiTokenCreateRequest(BaseModel): @@ -33,3 +35,5 @@ class ApiTokenCreateRequest(BaseModel): expires_in_days: int | None = Field(default=90, ge=0, le=3650) # Admin may create a token for another user; others ignored / forced to self. user_id: str | None = None + # Capability subset; empty inherits owner scopes. + scopes: list[str] | None = None diff --git a/netx_api/auth_scopes.py b/netx_api/auth_scopes.py new file mode 100644 index 0000000..21a2df5 --- /dev/null +++ b/netx_api/auth_scopes.py @@ -0,0 +1,178 @@ +"""Capability scopes for RBAC and API tokens. + +Roles map to default scopes; API tokens may further restrict via intersection. +""" + +from __future__ import annotations + +from typing import Iterable + +# Canonical scope names (keep stable for MCP / UI). +SCOPE_ALARMS_READ = "alarms:read" +SCOPE_NE_READ = "ne:read" +SCOPE_NE_WRITE = "ne:write" +SCOPE_NE_EXEC = "ne:exec" +SCOPE_WEBCRT = "webcrt:session" +SCOPE_SQL = "sql:query" +SCOPE_ADMIN_USERS = "admin:users" +SCOPE_OPS_WRITE = "ops:write" + +ALL_SCOPES: frozenset[str] = frozenset( + { + SCOPE_ALARMS_READ, + SCOPE_NE_READ, + SCOPE_NE_WRITE, + SCOPE_NE_EXEC, + SCOPE_WEBCRT, + SCOPE_SQL, + SCOPE_ADMIN_USERS, + SCOPE_OPS_WRITE, + } +) + +ROLE_DEFAULT_SCOPES: dict[str, frozenset[str]] = { + "admin": ALL_SCOPES, + # Read-only operator by default (alarms + inventory). + "user": frozenset({SCOPE_ALARMS_READ, SCOPE_NE_READ}), +} + +# Default MCP bootstrap token: diagnostics CLI allowed; no interactive shell / SQL / writes. +MCP_DEFAULT_SCOPES: tuple[str, ...] = ( + SCOPE_ALARMS_READ, + SCOPE_NE_READ, + SCOPE_NE_EXEC, +) + + +def normalize_scopes(raw: Iterable[str] | None) -> list[str]: + out: list[str] = [] + seen: set[str] = set() + for item in raw or []: + s = str(item or "").strip().lower() + if not s or s not in ALL_SCOPES or s in seen: + continue + seen.add(s) + out.append(s) + return sorted(out) + + +def scopes_for_role(role: str) -> frozenset[str]: + role_n = str(role or "user").strip().lower() + return ROLE_DEFAULT_SCOPES.get(role_n, ROLE_DEFAULT_SCOPES["user"]) + + +def effective_user_scopes(*, role: str, override: Iterable[str] | None = None) -> frozenset[str]: + """User scopes = role defaults, optionally replaced by a non-empty override list.""" + ov = normalize_scopes(override) + if ov: + # Admin role always keeps admin:users even if override omits it? No — override is authoritative + # when set; UI should only allow admins to set overrides. + return frozenset(ov) + return scopes_for_role(role) + + +def effective_token_scopes( + *, + user_scopes: Iterable[str], + token_scopes: Iterable[str] | None, +) -> frozenset[str]: + """Token cannot grant more than the owning user. Empty token scopes inherit user scopes.""" + user = frozenset(normalize_scopes(user_scopes)) + tok = normalize_scopes(token_scopes) + if not tok: + return user + return frozenset(tok) & user + + +def has_scope(granted: Iterable[str], required: str) -> bool: + need = str(required or "").strip().lower() + if not need: + return True + return need in frozenset(normalize_scopes(granted)) + + +def has_all_scopes(granted: Iterable[str], required: Iterable[str]) -> bool: + g = frozenset(normalize_scopes(granted)) + return all(str(r).strip().lower() in g for r in required if str(r).strip()) + + +def required_scope_for_request(method: str, path: str) -> str | None: + """Return a single required scope for the HTTP request, or None if any auth is enough. + + Paths already gated as public by middleware are not called here. + """ + m = (method or "GET").upper() + p = str(path or "") + + if p.startswith("/v1/users"): + return SCOPE_ADMIN_USERS + + if p.startswith("/v1/sql"): + return SCOPE_SQL + + if p.startswith("/v1/webcrt"): + return SCOPE_WEBCRT + + if p.startswith("/v1/managed-ne"): + if p.rstrip("/").endswith("/exec") and m == "POST": + return SCOPE_NE_EXEC + if m in ("POST", "PUT", "PATCH", "DELETE"): + return SCOPE_NE_WRITE + return SCOPE_NE_READ + + if p.startswith("/v1/cli"): + if m in ("POST", "PUT", "PATCH", "DELETE"): + return SCOPE_NE_WRITE + return SCOPE_NE_READ + + if p.startswith("/v1/ne-collections"): + if m in ("POST", "PUT", "PATCH", "DELETE"): + return SCOPE_NE_WRITE + return SCOPE_NE_READ + + if p.startswith("/v1/config-sync"): + if m in ("POST", "PUT", "PATCH", "DELETE"): + return SCOPE_NE_WRITE + return SCOPE_NE_READ + + if p.startswith("/v1/port-traffic"): + if m in ("POST", "PUT", "PATCH", "DELETE"): + return SCOPE_NE_WRITE + return SCOPE_NE_READ + + if p.startswith("/v1/topology"): + if m in ("POST", "PUT", "PATCH", "DELETE"): + return SCOPE_NE_WRITE + return SCOPE_NE_READ + + # UME / alarms / diagnostics / AI analyze / import + if ( + p.startswith("/v1/ume") + or p.startswith("/v1/alarms") + or p.startswith("/v1/batches") + or p.startswith("/v1/diagnostics") + or p.startswith("/v1/ap/") + or p.startswith("/v1/import") + or p.startswith("/v1/key-alert") + ): + if m in ("POST", "PUT", "PATCH", "DELETE") and ( + "/runtime/" in p or p.endswith("/pause") or p.endswith("/resume") or "subscription" in p + ): + return SCOPE_OPS_WRITE + if m in ("POST", "PUT", "PATCH", "DELETE") and not p.startswith("/v1/ap/"): + # Sync triggers / rule edits need ops write; pure analyze stays read. + if "/analyze" in p: + return SCOPE_ALARMS_READ + return SCOPE_OPS_WRITE + return SCOPE_ALARMS_READ + + if p.startswith("/v1/ops"): + if m in ("POST", "PUT", "PATCH", "DELETE"): + return SCOPE_OPS_WRITE + return SCOPE_NE_READ + + if p.startswith("/v1/integrations"): + return SCOPE_ALARMS_READ + + # Auth self-service, api-tokens, audit: any authenticated user + return None diff --git a/netx_api/auth_service.py b/netx_api/auth_service.py index e792314..32fee55 100644 --- a/netx_api/auth_service.py +++ b/netx_api/auth_service.py @@ -13,6 +13,11 @@ from sqlalchemy import func from sqlalchemy.orm import Session from .auth_passwords import hash_password, verify_password +from .auth_scopes import ( + MCP_DEFAULT_SCOPES, + effective_user_scopes, + normalize_scopes, +) from .auth_tokens import hash_api_token, issue_access_token, new_api_token_plaintext from .config import settings from .models import ApiToken, AppUser, AuditLog @@ -36,10 +41,15 @@ _SECRET_KEYS = frozenset( def user_public(user: AppUser) -> dict[str, Any]: + scopes = sorted( + effective_user_scopes(role=str(user.role or "user"), override=getattr(user, "scopes", None) or []) + ) return { "id": user.id, "username": user.username, "role": user.role, + "scopes": scopes, + "scopes_override": normalize_scopes(getattr(user, "scopes", None) or []), "is_active": bool(user.is_active), "must_change_password": bool(getattr(user, "must_change_password", False)), "created_by": user.created_by or "", @@ -177,6 +187,15 @@ def ensure_default_mcp_token(db: Session, user: AppUser | None = None) -> str | .one_or_none() ) if row is not None: + # Ensure MCP bootstrap token stays within MCP_DEFAULT_SCOPES. + desired = normalize_scopes(MCP_DEFAULT_SCOPES) + current = normalize_scopes(getattr(row, "scopes", None) or []) + if current != desired: + row.scopes = desired + try: + db.commit() + except Exception: + db.rollback() return existing except Exception: _log.exception("read mcp token file failed path=%s", path) @@ -192,7 +211,13 @@ def ensure_default_mcp_token(db: Session, user: AppUser | None = None) -> str | if admin is None: return None try: - row, plaintext = create_api_token(db, user=admin, name="mcp-default", expires_in_days=0) + row, plaintext = create_api_token( + db, + user=admin, + name="mcp-default", + expires_in_days=0, + scopes=list(MCP_DEFAULT_SCOPES), + ) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(plaintext + "\n", encoding="utf-8") try: @@ -251,6 +276,7 @@ def create_user( password: str, role: str, actor: AppUser, + scopes: list[str] | None = None, ) -> AppUser: name = str(username or "").strip() if not _USERNAME_RE.match(name): @@ -263,10 +289,12 @@ def create_user( raise HTTPException(status_code=400, detail="invalid_role") if get_user_by_username(db, name) is not None: raise HTTPException(status_code=409, detail="username_exists") + scope_list = normalize_scopes(scopes) if scopes is not None else [] user = AppUser( username=name, password_hash=hash_password(pwd), role=role_n, + scopes=scope_list, is_active=True, created_by=actor.id, ) @@ -284,6 +312,7 @@ def update_user( is_active: bool | None = None, role: str | None = None, password: str | None = None, + scopes: list[str] | None = None, ) -> AppUser: user = get_user_by_id(db, user_id) if user is None: @@ -305,6 +334,8 @@ def update_user( raise HTTPException(status_code=400, detail="password_too_short") user.password_hash = hash_password(pwd) user.must_change_password = True + if scopes is not None: + user.scopes = normalize_scopes(scopes) user.updated_at = datetime.utcnow() db.commit() db.refresh(user) @@ -312,7 +343,8 @@ def update_user( def change_password(db: Session, *, user: AppUser, old_password: str, new_password: str) -> None: - if not verify_password(old_password, user.password_hash): + row = get_user_by_id(db, str(user.id)) or user + if not verify_password(old_password, row.password_hash): raise HTTPException(status_code=400, detail="old_password_incorrect") pwd = str(new_password or "") if len(pwd) < 6: @@ -320,9 +352,9 @@ def change_password(db: Session, *, user: AppUser, old_password: str, new_passwo default_pwd = str(settings.bootstrap_admin_password or "admin123").strip() or "admin123" if pwd == default_pwd or pwd == old_password: raise HTTPException(status_code=400, detail="password_must_differ_from_default") - user.password_hash = hash_password(pwd) - user.must_change_password = False - user.updated_at = datetime.utcnow() + row.password_hash = hash_password(pwd) + row.must_change_password = False + row.updated_at = datetime.utcnow() db.commit() @@ -332,6 +364,7 @@ def create_api_token( user: AppUser, name: str, expires_in_days: int | None = None, + scopes: list[str] | None = None, ) -> tuple[ApiToken, str]: label = str(name or "").strip() or "default" if len(label) > 128: @@ -340,10 +373,16 @@ def create_api_token( if expires_in_days is not None and int(expires_in_days) > 0: expires_at = datetime.utcnow() + timedelta(days=int(expires_in_days)) plaintext = new_api_token_plaintext() + scope_list = normalize_scopes(scopes) if scopes is not None else [] + # Cap token scopes to owner's effective scopes. + owner_scopes = effective_user_scopes(role=str(user.role or "user"), override=getattr(user, "scopes", None) or []) + if scope_list: + scope_list = sorted(frozenset(scope_list) & owner_scopes) row = ApiToken( name=label, token_hash=hash_api_token(plaintext), user_id=user.id, + scopes=scope_list, expires_at=expires_at, ) db.add(row) @@ -361,6 +400,7 @@ def _token_public(db: Session, r: ApiToken) -> dict[str, Any]: "name": r.name, "user_id": r.user_id, "username": owner.username if owner else "", + "scopes": normalize_scopes(getattr(r, "scopes", None) or []), "created_at": r.created_at.isoformat() if r.created_at else None, "expires_at": r.expires_at.isoformat() if r.expires_at else None, "last_used_at": r.last_used_at.isoformat() if r.last_used_at else None, @@ -392,7 +432,7 @@ def revoke_api_token(db: Session, *, token_id: str, actor: AppUser) -> ApiToken: return row -def resolve_api_token_user(db: Session, plaintext: str) -> AppUser | None: +def resolve_api_token_row(db: Session, plaintext: str) -> ApiToken | None: th = hash_api_token(plaintext) row = ( db.query(ApiToken) @@ -411,7 +451,14 @@ def resolve_api_token_user(db: Session, plaintext: str) -> AppUser | None: db.commit() except Exception: db.rollback() - return user + return row + + +def resolve_api_token_user(db: Session, plaintext: str) -> AppUser | None: + row = resolve_api_token_row(db, plaintext) + if row is None: + return None + return get_user_by_id(db, row.user_id) def list_audit_logs( diff --git a/netx_api/auth_tokens.py b/netx_api/auth_tokens.py index 7f76790..0ca3d28 100644 --- a/netx_api/auth_tokens.py +++ b/netx_api/auth_tokens.py @@ -6,6 +6,7 @@ import hashlib import logging import secrets from datetime import datetime, timedelta, timezone +from pathlib import Path from typing import Any import jwt @@ -14,20 +15,70 @@ from .config import settings _log = logging.getLogger("netx.auth") -_DEFAULT_DEV_SECRET = "netx-dev-auth-secret-change-me-in-production-32b" -_warned_default_secret = False +# Legacy hard-coded value (pre auto-file). Treated as insecure if still set via env. +_LEGACY_INSECURE_SECRET = "netx-dev-auth-secret-change-me-in-production-32b" +_cached_secret: str | None = None + + +def auth_secret_file_path() -> Path: + raw = str(getattr(settings, "auth_secret_file", None) or "data/auth/jwt_secret").strip() + path = Path(raw) + if not path.is_absolute(): + path = Path.cwd() / path + return path + + +def is_legacy_insecure_secret(value: str) -> bool: + return str(value or "").strip() == _LEGACY_INSECURE_SECRET + + +def ensure_auth_secret() -> str: + """Resolve JWT signing secret: explicit env > persisted file > generate once. + + Empty / legacy built-in ``NETX_AUTH_SECRET`` falls through to a per-install + file under ``data/auth/jwt_secret`` (same idea as ``mcp_token``). + """ + global _cached_secret + configured = str(settings.auth_secret or "").strip() + if configured and not is_legacy_insecure_secret(configured): + return configured + if configured and is_legacy_insecure_secret(configured): + _log.warning( + "NETX_AUTH_SECRET is the legacy shared default; ignoring it and using " + "per-install file %s (existing JWTs will need re-login)", + auth_secret_file_path(), + ) + if _cached_secret: + return _cached_secret + + path = auth_secret_file_path() + try: + if path.is_file(): + existing = path.read_text(encoding="utf-8").strip() + if len(existing) >= 32 and not is_legacy_insecure_secret(existing): + _cached_secret = existing + return _cached_secret + except Exception: + _log.exception("read auth secret file failed path=%s", path) + + generated = secrets.token_urlsafe(48) + try: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(generated + "\n", encoding="utf-8") + try: + path.chmod(0o600) + except Exception: + pass + _log.info("wrote per-install JWT secret to %s", path) + except Exception: + _log.exception("write auth secret file failed path=%s; using in-memory secret only", path) + _cached_secret = generated + return _cached_secret def auth_secret() -> str: - """Return configured secret (lab default is set in Settings).""" - global _warned_default_secret - configured = str(settings.auth_secret or "").strip() or _DEFAULT_DEV_SECRET - if configured == _DEFAULT_DEV_SECRET and not _warned_default_secret: - _warned_default_secret = True - _log.warning( - "using default NETX_AUTH_SECRET; set a unique secret for production deployments" - ) - return configured + """Return JWT HMAC secret (ensures file exists on first use).""" + return ensure_auth_secret() def issue_access_token(*, user_id: str, username: str, role: str) -> str: diff --git a/netx_api/config.py b/netx_api/config.py index b0024eb..7a31f9c 100644 --- a/netx_api/config.py +++ b/netx_api/config.py @@ -26,7 +26,7 @@ class Settings(BaseSettings): ume_base_url: str = "" ume_username: str = "" ume_password: str = "" - ume_verify_tls: bool = False + ume_verify_tls: bool = True ume_timeout_s: float = 20.0 ume_page_size: int = 1000 ume_max_pages: int = 2000 @@ -117,13 +117,28 @@ class Settings(BaseSettings): webcrt_sftp_list_timeout_sec: float = 30.0 # Local app login / audit (lab defaults; override in production) auth_enabled: bool = True - # Stable default so JWT survives restarts without .env. Override in production. - auth_secret: str = "netx-dev-auth-secret-change-me-in-production-32b" + # Empty = auto-generate & persist to auth_secret_file (recommended). + # Set explicitly only when you want a shared/ops-managed secret. + auth_secret: str = "" + auth_secret_file: str = "data/auth/jwt_secret" auth_token_ttl_sec: int = 86400 bootstrap_admin_username: str = "admin" bootstrap_admin_password: str = "admin123" # Written on first boot for MCP; path relative to cwd / absolute auth_mcp_token_file: str = "data/auth/mcp_token" + # Expose /docs /redoc /openapi.json without auth when true (lab only). + docs_enabled: bool = False + # Refuse start when bind host is non-loopback and insecure defaults remain. + allow_insecure_defaults: bool = False + # Async audit writer; sample_n>1 keeps 1/N of generic http.* events. + audit_async: bool = True + audit_sample_n: int = 1 + # Prefer Alembic; when true, skip ad-hoc ALTER TABLE on startup (create_all still runs). + skip_legacy_startup_ddl: bool = False + # Optional dedicated SQLAlchemy URL for /v1/sql/* (read-only DB role recommended). + sql_readonly_database_url: str = "" + # When false, API skips config_sync / lldp / port_traffic schedulers (run `python -m netx_api.worker`). + run_inline_schedulers: bool = True settings = Settings() diff --git a/netx_api/integrations_router.py b/netx_api/integrations_router.py new file mode 100644 index 0000000..fa1430e --- /dev/null +++ b/netx_api/integrations_router.py @@ -0,0 +1,29 @@ +"""Thin integrations / health routes (liveness vs readiness).""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends +from sqlalchemy import text as sql_text +from sqlalchemy.orm import Session + +from .db import get_db + +router = APIRouter(tags=["health"]) + + +@router.get("/health/live", status_code=200) +def health_live() -> dict[str, str]: + """Process liveness — no DB or upstream checks.""" + return {"status": "ok", "probe": "live"} + + +@router.get("/health/ready", status_code=200) +def health_ready(db: Session = Depends(get_db)) -> dict[str, Any]: + """Readiness — verifies database connectivity.""" + try: + db.execute(sql_text("select 1")) + return {"status": "ok", "probe": "ready", "db": "up"} + except Exception as exc: + return {"status": "down", "probe": "ready", "db": "down", "error": str(exc)[:240]} diff --git a/netx_api/main.py b/netx_api/main.py index fbd6c3d..5d7f527 100644 --- a/netx_api/main.py +++ b/netx_api/main.py @@ -33,6 +33,11 @@ from .webcrt_router import router as webcrt_router from .topology_router import router as topology_router from .lldp_collect_router import router as lldp_collect_router from .ops_router import router as ops_router +from .sql_router import router as sql_router +from .sql_router import sql_query, sql_ume_query # noqa: F401 — tests import from main +from .security_bootstrap import assert_secure_defaults_or_exit +from .integrations_router import router as integrations_router +from .ume_runtime import start_device_schedulers from .importer import aggregate_alarms, import_alarm_excel, query_alarms from .models import ( AiAnalyzeHistory, @@ -131,7 +136,13 @@ from .schemas import ( ImportJobListResponse, ) -app = FastAPI(title="netx ops tool", version="0.1.0") +app = FastAPI( + title="netx ops tool", + version="0.1.0", + docs_url="/docs" if bool(settings.docs_enabled) else None, + redoc_url="/redoc" if bool(settings.docs_enabled) else None, + openapi_url="/openapi.json" if bool(settings.docs_enabled) else None, +) app.add_middleware(AuthAuditMiddleware) app.include_router(auth_router) app.include_router(managed_ne_router) @@ -143,6 +154,8 @@ app.include_router(webcrt_router) app.include_router(topology_router) app.include_router(lldp_collect_router) app.include_router(ops_router) +app.include_router(sql_router) +app.include_router(integrations_router) parser_cfg = load_parser_config() _UME_CLIENT_SINGLETON = UMEClient( token_loader=lambda: load_shared_token(), @@ -621,115 +634,6 @@ def _ume_error_kind(err: str) -> str: return "other" -@app.post("/v1/sql/query") -def sql_query(payload: dict[str, Any] | None = None, db: Session = Depends(get_db)) -> dict: - """ - Read-only SQL query endpoint for AI power users. - - Safety constraints: - - SELECT only, single statement (no ';') - - forbid DDL/DML keywords - - enforce max rows (server-side LIMIT wrapper) - - require batch_id param and require SQL contains ':batch_id' - """ - payload = payload or {} - sql = str(payload.get("sql") or "").strip() - batch_id = str(payload.get("batch_id") or "").strip() - limit = int(payload.get("limit") or 200) - limit = max(1, min(limit, 2000)) - if not sql: - raise HTTPException(status_code=400, detail="sql_required") - if ";" in sql: - raise HTTPException(status_code=400, detail="single_statement_only") - low = sql.lower().lstrip() - if not low.startswith("select"): - raise HTTPException(status_code=400, detail="select_only") - if _SQL_FORBIDDEN_RE.search(sql): - raise HTTPException(status_code=400, detail="forbidden_keyword") - if not batch_id: - raise HTTPException(status_code=400, detail="batch_id_required") - if ":batch_id" not in sql: - raise HTTPException(status_code=400, detail="batch_id_param_required(:batch_id)") - wrapped = f"select * from ({sql}) as q limit {limit}" - try: - res = db.execute(sql_text(wrapped), {"batch_id": batch_id}) - cols = list(res.keys()) - raw_rows = res.fetchall() - rows: list[list[Any]] = [] - for r in raw_rows: - out_row: list[Any] = [] - for v in list(r): - if isinstance(v, datetime): - out_row.append(((_ensure_utc(v) or v).isoformat().replace("+00:00", "Z"))) - else: - out_row.append(v) - rows.append(out_row) - return {"ok": True, "columns": cols, "rows": rows, "limit": limit} - except Exception as exc: - raise HTTPException(status_code=400, detail=f"sql_failed:{str(exc)[:240]}") from exc - - -@app.post("/v1/sql/ume_query") -def sql_ume_query(payload: dict[str, Any] | None = None, db: Session = Depends(get_db)) -> dict: - """ - Read-only SQL query endpoint for UME current alarms/inventory. - - Safety constraints: - - SELECT only, single statement (no ';') - - forbid DDL/DML keywords - - enforce max rows (server-side LIMIT wrapper) - - only allow FROM/JOIN on ume_alarms_current and ume_inventory_ne - """ - payload = payload or {} - sql = str(payload.get("sql") or "").strip() - limit = int(payload.get("limit") or 200) - limit = max(1, min(limit, 2000)) - statement_timeout_ms = int(payload.get("statement_timeout_ms") or 0) - statement_timeout_ms = max(0, min(statement_timeout_ms, 30000)) - if not sql: - raise HTTPException(status_code=400, detail="sql_required") - if ";" in sql: - raise HTTPException(status_code=400, detail="single_statement_only") - low = sql.lower().lstrip() - if not low.startswith("select"): - raise HTTPException(status_code=400, detail="select_only") - if _SQL_FORBIDDEN_RE.search(sql): - raise HTTPException(status_code=400, detail="forbidden_keyword") - - allowed_tables = {"ume_alarms_current", "ume_inventory_ne"} - refs = re.findall(r"\b(?:from|join)\s+([a-zA-Z0-9_\"\.]+)", sql, flags=re.IGNORECASE) - for ref in refs: - normalized = str(ref).strip().strip('"') - if "." in normalized: - normalized = normalized.split(".")[-1] - if normalized.lower() not in allowed_tables: - raise HTTPException(status_code=400, detail=f"ume_table_not_allowed:{normalized}") - - wrapped = f"select * from ({sql}) as q limit {limit}" - try: - if statement_timeout_ms > 0: - try: - if str(getattr(getattr(db, "bind", None), "dialect", None).name).lower().startswith("postgres"): - db.execute(sql_text("SET LOCAL statement_timeout = :ms"), {"ms": int(statement_timeout_ms)}) - except Exception: - pass - res = db.execute(sql_text(wrapped)) - cols = list(res.keys()) - raw_rows = res.fetchall() - rows: list[list[Any]] = [] - for r in raw_rows: - out_row: list[Any] = [] - for v in list(r): - if isinstance(v, datetime): - out_row.append(((_ensure_utc(v) or v).isoformat().replace("+00:00", "Z"))) - else: - out_row.append(v) - rows.append(out_row) - return {"ok": True, "columns": cols, "rows": rows, "limit": limit} - except Exception as exc: - raise HTTPException(status_code=400, detail=f"sql_failed:{str(exc)[:240]}") from exc - - def _configure_ume_diag_logging() -> None: """Emit netx.ume.* INFO to stderr so background scripts/.run/*.log and consoles show scheduler lines.""" fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s") @@ -807,33 +711,50 @@ def _migrate_key_alert_rule_schema() -> None: @app.on_event("startup") def on_startup() -> None: + assert_secure_defaults_or_exit() _configure_ume_diag_logging() Base.metadata.create_all(bind=engine) - _migrate_key_alert_rule_schema() + skip_ddl = bool(getattr(settings, "skip_legacy_startup_ddl", False)) + if not skip_ddl: + _migrate_key_alert_rule_schema() # Auth columns must exist before bootstrap / flag_default_password_users. try: with engine.begin() as conn: conn.exec_driver_sql( "ALTER TABLE app_user ADD COLUMN IF NOT EXISTS must_change_password BOOLEAN DEFAULT FALSE" ) - conn.exec_driver_sql("ALTER TABLE api_token ADD COLUMN IF NOT EXISTS expires_at TIMESTAMP") - from .port_traffic_migrate import ensure_port_traffic_series_schema - from .topology_migrate import ensure_topology_schema - - ensure_port_traffic_series_schema(conn) - ensure_topology_schema(conn) - conn.exec_driver_sql( - "ALTER TABLE ne_collection_run ADD COLUMN IF NOT EXISTS ne_source VARCHAR(16) DEFAULT 'managed'" - ) + # JSON works on Postgres/SQLite; create_all also defines ORM column. try: conn.exec_driver_sql( - "ALTER TABLE ne_collection_run ALTER COLUMN ne_id TYPE VARCHAR(128)" + "ALTER TABLE app_user ADD COLUMN IF NOT EXISTS scopes JSON DEFAULT '[]'" ) except Exception: pass - conn.exec_driver_sql( - "ALTER TABLE config_sync_policy ADD COLUMN IF NOT EXISTS cycle_keep INTEGER DEFAULT 30" - ) + conn.exec_driver_sql("ALTER TABLE api_token ADD COLUMN IF NOT EXISTS expires_at TIMESTAMP") + try: + conn.exec_driver_sql( + "ALTER TABLE api_token ADD COLUMN IF NOT EXISTS scopes JSON DEFAULT '[]'" + ) + except Exception: + pass + if not skip_ddl: + from .port_traffic_migrate import ensure_port_traffic_series_schema + from .topology_migrate import ensure_topology_schema + + ensure_port_traffic_series_schema(conn) + ensure_topology_schema(conn) + conn.exec_driver_sql( + "ALTER TABLE ne_collection_run ADD COLUMN IF NOT EXISTS ne_source VARCHAR(16) DEFAULT 'managed'" + ) + try: + conn.exec_driver_sql( + "ALTER TABLE ne_collection_run ALTER COLUMN ne_id TYPE VARCHAR(128)" + ) + except Exception: + pass + conn.exec_driver_sql( + "ALTER TABLE config_sync_policy ADD COLUMN IF NOT EXISTS cycle_keep INTEGER DEFAULT 30" + ) except Exception: _schedule_log.exception("startup: auth/port_traffic/topology schema migration failed") _reset_runtime_pause_flags() @@ -899,232 +820,228 @@ def on_startup() -> None: _schedule_log.exception("startup: ne collection / config_sync recovery failed") finally: db.close() - try: - from .config_sync_scheduler import start_config_sync_scheduler - - start_config_sync_scheduler() - except Exception: - _schedule_log.exception("startup: config_sync scheduler init failed") - try: - from .lldp_collect_scheduler import start_lldp_collect_scheduler - - start_lldp_collect_scheduler() - except Exception: - _schedule_log.exception("startup: lldp_collect scheduler init failed") - try: - from .port_traffic_scheduler import start_port_traffic_scheduler - - start_port_traffic_scheduler() - except Exception: - _schedule_log.exception("startup: port_traffic scheduler init failed") + if bool(getattr(settings, "run_inline_schedulers", True)): + try: + start_device_schedulers() + except Exception: + _schedule_log.exception("startup: device schedulers init failed") + else: + _schedule_log.info( + "startup: inline schedulers disabled — run `python -m netx_api.worker` for " + "config_sync / lldp_collect / port_traffic" + ) # Best-effort schema evolution for new columns (no migrations framework). # Safe for Postgres (IF NOT EXISTS); ignored on failure. - try: - with engine.begin() as conn: - # Removed from ORM: drop legacy holder table if present (was optional nested UME data). - conn.exec_driver_sql("DROP TABLE IF EXISTS ume_inventory_equipment_holder") - conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS relevancy VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS l3vpn_peer_ne VARCHAR(256) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS service VARCHAR(256) DEFAULT ''") - conn.exec_driver_sql( - "ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS affected_client_service_number INTEGER DEFAULT 0" - ) - conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS intermittence_count INTEGER DEFAULT 0") - conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS me_level VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_token_cache ADD COLUMN IF NOT EXISTS lock_owner VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_token_cache ADD COLUMN IF NOT EXISTS lock_expires_at_epoch_s INTEGER DEFAULT 0") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS device_level VARCHAR(64) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS host_name VARCHAR(256) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS location VARCHAR(512) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS ipv6_address VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql( - "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS hardware_version VARCHAR(128) DEFAULT ''" - ) - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS loopback VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql( - "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS consistent_state VARCHAR(64) DEFAULT ''" - ) - conn.exec_driver_sql( - "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS interface_version VARCHAR(128) DEFAULT ''" - ) - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS mac VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS admin_status VARCHAR(64) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS address_type VARCHAR(64) DEFAULT ''") - conn.exec_driver_sql( - "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS connection_status VARCHAR(64) DEFAULT ''" - ) - conn.exec_driver_sql( - "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS maintain_status VARCHAR(64) DEFAULT ''" - ) - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS net_mask VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS create_time VARCHAR(64) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS creator VARCHAR(128) DEFAULT ''") - # Allow long UME alarm fields; avoid StringDataRightTruncation on large payloads. - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN alarm_key TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN object_name TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN event_type TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN native_probable_cause TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN perceived_severity TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN is_cleared TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN time_created TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN root_cause_alarm_indication TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ADD COLUMN IF NOT EXISTS host_name VARCHAR(256) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ADD COLUMN IF NOT EXISTS host_name VARCHAR(256) DEFAULT ''") - conn.exec_driver_sql( - "ALTER TABLE ume_alarms_current ADD COLUMN IF NOT EXISTS notification_id VARCHAR(128) DEFAULT ''" - ) - conn.exec_driver_sql( - "ALTER TABLE ume_alarms_history ADD COLUMN IF NOT EXISTS notification_id VARCHAR(128) DEFAULT ''" - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_ume_alarms_current_notification_id ON ume_alarms_current (notification_id)" - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_ume_alarms_history_notification_id ON ume_alarms_history (notification_id)" - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_ume_alarms_current_host_name ON ume_alarms_current (host_name)" - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_ume_alarms_history_host_name ON ume_alarms_history (host_name)" - ) - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN alarm_key TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN object_name TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN event_type TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN native_probable_cause TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN perceived_severity TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN is_cleared TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN time_created TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN root_cause_alarm_indication TYPE TEXT") - # Simplify alarm tables: display fields come from runtime join with inventory table. - conn.exec_driver_sql("ALTER TABLE ume_alarms_current DROP COLUMN IF EXISTS ne_name") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current DROP COLUMN IF EXISTS user_label") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history DROP COLUMN IF EXISTS ne_name") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history DROP COLUMN IF EXISTS user_label") - conn.exec_driver_sql("ALTER TABLE api_token ADD COLUMN IF NOT EXISTS expires_at TIMESTAMP") - conn.exec_driver_sql( - "ALTER TABLE app_user ADD COLUMN IF NOT EXISTS must_change_password BOOLEAN DEFAULT FALSE" - ) - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_enabled BOOLEAN DEFAULT FALSE") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_vendor VARCHAR(32) DEFAULT 'zte'") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_host VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_port INTEGER DEFAULT 22") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_protocol VARCHAR(16) DEFAULT 'ssh'") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_username VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_password_enc TEXT DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_command_template TEXT DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_vrf VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql( - "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_target_auth_mode VARCHAR(32) DEFAULT 'bastion_managed'" - ) - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS source VARCHAR(64) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS source_ref VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS connect_detail TEXT DEFAULT ''") - # WebCRT sessions may share a host IP; uniqueness is enforced in ne_service for inventory only. - for stmt in ( - "ALTER TABLE managed_ne DROP CONSTRAINT IF EXISTS managed_ne_ip_address_key", - "DROP INDEX IF EXISTS managed_ne_ip_address_key", - "DROP INDEX IF EXISTS ix_managed_ne_ip_address", - "DROP INDEX IF EXISTS sqlite_autoindex_managed_ne_1", - ): + # When NETX_SKIP_LEGACY_STARTUP_DDL=1, rely on Alembic instead. + if bool(getattr(settings, "skip_legacy_startup_ddl", False)): + _schedule_log.info("startup: skipping legacy ALTER TABLE DDL (Alembic mode)") + else: + try: + with engine.begin() as conn: + # Removed from ORM: drop legacy holder table if present (was optional nested UME data). + conn.exec_driver_sql("DROP TABLE IF EXISTS ume_inventory_equipment_holder") + conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS relevancy VARCHAR(128) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS l3vpn_peer_ne VARCHAR(256) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS service VARCHAR(256) DEFAULT ''") + conn.exec_driver_sql( + "ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS affected_client_service_number INTEGER DEFAULT 0" + ) + conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS intermittence_count INTEGER DEFAULT 0") + conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS me_level VARCHAR(128) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE ume_token_cache ADD COLUMN IF NOT EXISTS lock_owner VARCHAR(128) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE ume_token_cache ADD COLUMN IF NOT EXISTS lock_expires_at_epoch_s INTEGER DEFAULT 0") + conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS device_level VARCHAR(64) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS host_name VARCHAR(256) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS location VARCHAR(512) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS ipv6_address VARCHAR(128) DEFAULT ''") + conn.exec_driver_sql( + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS hardware_version VARCHAR(128) DEFAULT ''" + ) + conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS loopback VARCHAR(128) DEFAULT ''") + conn.exec_driver_sql( + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS consistent_state VARCHAR(64) DEFAULT ''" + ) + conn.exec_driver_sql( + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS interface_version VARCHAR(128) DEFAULT ''" + ) + conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS mac VARCHAR(128) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS admin_status VARCHAR(64) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS address_type VARCHAR(64) DEFAULT ''") + conn.exec_driver_sql( + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS connection_status VARCHAR(64) DEFAULT ''" + ) + conn.exec_driver_sql( + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS maintain_status VARCHAR(64) DEFAULT ''" + ) + conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS net_mask VARCHAR(128) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS create_time VARCHAR(64) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS creator VARCHAR(128) DEFAULT ''") + # Allow long UME alarm fields; avoid StringDataRightTruncation on large payloads. + conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN alarm_key TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN object_name TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN event_type TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN native_probable_cause TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN perceived_severity TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN is_cleared TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN time_created TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN root_cause_alarm_indication TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_current ADD COLUMN IF NOT EXISTS host_name VARCHAR(256) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE ume_alarms_history ADD COLUMN IF NOT EXISTS host_name VARCHAR(256) DEFAULT ''") + conn.exec_driver_sql( + "ALTER TABLE ume_alarms_current ADD COLUMN IF NOT EXISTS notification_id VARCHAR(128) DEFAULT ''" + ) + conn.exec_driver_sql( + "ALTER TABLE ume_alarms_history ADD COLUMN IF NOT EXISTS notification_id VARCHAR(128) DEFAULT ''" + ) + conn.exec_driver_sql( + "CREATE INDEX IF NOT EXISTS ix_ume_alarms_current_notification_id ON ume_alarms_current (notification_id)" + ) + conn.exec_driver_sql( + "CREATE INDEX IF NOT EXISTS ix_ume_alarms_history_notification_id ON ume_alarms_history (notification_id)" + ) + conn.exec_driver_sql( + "CREATE INDEX IF NOT EXISTS ix_ume_alarms_current_host_name ON ume_alarms_current (host_name)" + ) + conn.exec_driver_sql( + "CREATE INDEX IF NOT EXISTS ix_ume_alarms_history_host_name ON ume_alarms_history (host_name)" + ) + conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN alarm_key TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN object_name TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN event_type TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN native_probable_cause TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN perceived_severity TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN is_cleared TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN time_created TYPE TEXT") + conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN root_cause_alarm_indication TYPE TEXT") + # Simplify alarm tables: display fields come from runtime join with inventory table. + conn.exec_driver_sql("ALTER TABLE ume_alarms_current DROP COLUMN IF EXISTS ne_name") + conn.exec_driver_sql("ALTER TABLE ume_alarms_current DROP COLUMN IF EXISTS user_label") + conn.exec_driver_sql("ALTER TABLE ume_alarms_history DROP COLUMN IF EXISTS ne_name") + conn.exec_driver_sql("ALTER TABLE ume_alarms_history DROP COLUMN IF EXISTS user_label") + conn.exec_driver_sql("ALTER TABLE api_token ADD COLUMN IF NOT EXISTS expires_at TIMESTAMP") + conn.exec_driver_sql( + "ALTER TABLE app_user ADD COLUMN IF NOT EXISTS must_change_password BOOLEAN DEFAULT FALSE" + ) + conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_enabled BOOLEAN DEFAULT FALSE") + conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_vendor VARCHAR(32) DEFAULT 'zte'") + conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_host VARCHAR(128) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_port INTEGER DEFAULT 22") + conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_protocol VARCHAR(16) DEFAULT 'ssh'") + conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_username VARCHAR(128) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_password_enc TEXT DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_command_template TEXT DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_vrf VARCHAR(128) DEFAULT ''") + conn.exec_driver_sql( + "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_target_auth_mode VARCHAR(32) DEFAULT 'bastion_managed'" + ) + conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS source VARCHAR(64) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS source_ref VARCHAR(128) DEFAULT ''") + conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS connect_detail TEXT DEFAULT ''") + # WebCRT sessions may share a host IP; uniqueness is enforced in ne_service for inventory only. + for stmt in ( + "ALTER TABLE managed_ne DROP CONSTRAINT IF EXISTS managed_ne_ip_address_key", + "DROP INDEX IF EXISTS managed_ne_ip_address_key", + "DROP INDEX IF EXISTS ix_managed_ne_ip_address", + "DROP INDEX IF EXISTS sqlite_autoindex_managed_ne_1", + ): + try: + conn.exec_driver_sql(stmt) + except Exception: + pass try: - conn.exec_driver_sql(stmt) + conn.exec_driver_sql( + "CREATE INDEX IF NOT EXISTS ix_managed_ne_ip_address ON managed_ne (ip_address)" + ) except Exception: pass - try: conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_managed_ne_ip_address ON managed_ne (ip_address)" + "ALTER TABLE ne_collection_job ADD COLUMN IF NOT EXISTS last_run_at TIMESTAMP" ) - except Exception: - pass - conn.exec_driver_sql( - "ALTER TABLE ne_collection_job ADD COLUMN IF NOT EXISTS last_run_at TIMESTAMP" - ) - conn.exec_driver_sql( - "UPDATE ne_collection_job SET last_run_at = COALESCE(ended_at, started_at, created_at) " - "WHERE last_run_at IS NULL" - ) - conn.exec_driver_sql( - """ - CREATE TABLE IF NOT EXISTS cli_connect_profile ( - id VARCHAR(64) PRIMARY KEY, - name VARCHAR(256) DEFAULT '', - is_default BOOLEAN DEFAULT FALSE, - username VARCHAR(128) DEFAULT '', - password_enc TEXT DEFAULT '', - port INTEGER DEFAULT 22, - protocol VARCHAR(16) DEFAULT 'ssh', - device_type_default VARCHAR(128) DEFAULT 'zte_zxros', - vendor_default VARCHAR(64) DEFAULT 'ZTE', - ne_type_rules TEXT DEFAULT '', - hop_enabled BOOLEAN DEFAULT FALSE, - hop_vendor VARCHAR(32) DEFAULT 'zte', - hop_host VARCHAR(128) DEFAULT '', - hop_port INTEGER DEFAULT 22, - hop_protocol VARCHAR(16) DEFAULT 'ssh', - hop_username VARCHAR(128) DEFAULT '', - hop_password_enc TEXT DEFAULT '', - hop_command_template TEXT DEFAULT '', - hop_vrf VARCHAR(128) DEFAULT '', - hop_target_auth_mode VARCHAR(32) DEFAULT 'bastion_managed', - created_at TIMESTAMP, - updated_at TIMESTAMP + conn.exec_driver_sql( + "UPDATE ne_collection_job SET last_run_at = COALESCE(ended_at, started_at, created_at) " + "WHERE last_run_at IS NULL" ) - """ - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_cli_connect_profile_is_default ON cli_connect_profile (is_default)" - ) - conn.exec_driver_sql( - """ - CREATE TABLE IF NOT EXISTS ume_cli_override ( - ume_ne_id VARCHAR(128) PRIMARY KEY, - profile_id VARCHAR(64), - username_override VARCHAR(128) DEFAULT '', - device_type_override VARCHAR(128) DEFAULT '', - vendor_override VARCHAR(64) DEFAULT '', - connect_status VARCHAR(32) DEFAULT 'unknown', - connect_message VARCHAR(512) DEFAULT '', - connect_detail TEXT DEFAULT '', - connect_tested_at TIMESTAMP, - updated_at TIMESTAMP + conn.exec_driver_sql( + """ + CREATE TABLE IF NOT EXISTS cli_connect_profile ( + id VARCHAR(64) PRIMARY KEY, + name VARCHAR(256) DEFAULT '', + is_default BOOLEAN DEFAULT FALSE, + username VARCHAR(128) DEFAULT '', + password_enc TEXT DEFAULT '', + port INTEGER DEFAULT 22, + protocol VARCHAR(16) DEFAULT 'ssh', + device_type_default VARCHAR(128) DEFAULT 'zte_zxros', + vendor_default VARCHAR(64) DEFAULT 'ZTE', + ne_type_rules TEXT DEFAULT '', + hop_enabled BOOLEAN DEFAULT FALSE, + hop_vendor VARCHAR(32) DEFAULT 'zte', + hop_host VARCHAR(128) DEFAULT '', + hop_port INTEGER DEFAULT 22, + hop_protocol VARCHAR(16) DEFAULT 'ssh', + hop_username VARCHAR(128) DEFAULT '', + hop_password_enc TEXT DEFAULT '', + hop_command_template TEXT DEFAULT '', + hop_vrf VARCHAR(128) DEFAULT '', + hop_target_auth_mode VARCHAR(32) DEFAULT 'bastion_managed', + created_at TIMESTAMP, + updated_at TIMESTAMP + ) + """ ) - """ - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_ume_cli_override_connect_status ON ume_cli_override (connect_status)" - ) - conn.exec_driver_sql("COMMENT ON TABLE ume_inventory_ne IS '网元对象详细信息'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ne_id IS '网元uuid'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ne_name IS '资源名称'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ne_type IS '网元类型'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.user_label IS '用户标签'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.address_type IS '管理地址类型(1:IPv4,2:IPv6)'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ip_address IS '网元IPv4地址'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.net_mask IS '管理IPv4掩码(点分十进制)'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ipv6_address IS 'IPv6地址'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.admin_status IS '管理状态(0-离线,1-在线)'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.connection_status IS '连接状态(0-断链,1-正常)'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.consistent_state IS '数据一致性状态(1一致,2不一致,3冲突)'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.maintain_status IS '工程状态(0普通,1调测,2新建)'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.vendor IS '网元提供商'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.interface_version IS '网元接口版本号'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.hardware_version IS '硬件版本'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.mac IS '设备机架MAC地址'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.loopback IS '业务环回IP(IPv4)'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.device_level IS '网元层次'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.host_name IS '主机名称'") - conn.exec_driver_sql( - "ALTER TABLE topology_edge ADD COLUMN IF NOT EXISTS stroke_color VARCHAR(32) DEFAULT ''" - ) - conn.exec_driver_sql( - "ALTER TABLE topology_edge ADD COLUMN IF NOT EXISTS stroke_width INTEGER DEFAULT 0" - ) - conn.exec_driver_sql( - "ALTER TABLE topology_edge ADD COLUMN IF NOT EXISTS line_style VARCHAR(16) DEFAULT ''" - ) - except Exception: - pass + conn.exec_driver_sql( + "CREATE INDEX IF NOT EXISTS ix_cli_connect_profile_is_default ON cli_connect_profile (is_default)" + ) + conn.exec_driver_sql( + """ + CREATE TABLE IF NOT EXISTS ume_cli_override ( + ume_ne_id VARCHAR(128) PRIMARY KEY, + profile_id VARCHAR(64), + username_override VARCHAR(128) DEFAULT '', + device_type_override VARCHAR(128) DEFAULT '', + vendor_override VARCHAR(64) DEFAULT '', + connect_status VARCHAR(32) DEFAULT 'unknown', + connect_message VARCHAR(512) DEFAULT '', + connect_detail TEXT DEFAULT '', + connect_tested_at TIMESTAMP, + updated_at TIMESTAMP + ) + """ + ) + conn.exec_driver_sql( + "CREATE INDEX IF NOT EXISTS ix_ume_cli_override_connect_status ON ume_cli_override (connect_status)" + ) + conn.exec_driver_sql("COMMENT ON TABLE ume_inventory_ne IS '网元对象详细信息'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ne_id IS '网元uuid'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ne_name IS '资源名称'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ne_type IS '网元类型'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.user_label IS '用户标签'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.address_type IS '管理地址类型(1:IPv4,2:IPv6)'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ip_address IS '网元IPv4地址'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.net_mask IS '管理IPv4掩码(点分十进制)'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ipv6_address IS 'IPv6地址'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.admin_status IS '管理状态(0-离线,1-在线)'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.connection_status IS '连接状态(0-断链,1-正常)'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.consistent_state IS '数据一致性状态(1一致,2不一致,3冲突)'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.maintain_status IS '工程状态(0普通,1调测,2新建)'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.vendor IS '网元提供商'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.interface_version IS '网元接口版本号'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.hardware_version IS '硬件版本'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.mac IS '设备机架MAC地址'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.loopback IS '业务环回IP(IPv4)'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.device_level IS '网元层次'") + conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.host_name IS '主机名称'") + conn.exec_driver_sql( + "ALTER TABLE topology_edge ADD COLUMN IF NOT EXISTS stroke_color VARCHAR(32) DEFAULT ''" + ) + conn.exec_driver_sql( + "ALTER TABLE topology_edge ADD COLUMN IF NOT EXISTS stroke_width INTEGER DEFAULT 0" + ) + conn.exec_driver_sql( + "ALTER TABLE topology_edge ADD COLUMN IF NOT EXISTS line_style VARCHAR(16) DEFAULT ''" + ) + except Exception: + pass try: if bool(getattr(settings, "ume_keepalive_enabled", True)): interval_keepalive_s = int(getattr(settings, "ume_keepalive_interval_s", 600) or 600) @@ -1435,6 +1352,7 @@ def health() -> dict[str, str]: return {"status": "ok"} + @app.get("/v1/ume/token/status") def ume_token_status() -> dict[str, Any]: client = _ume_client() diff --git a/netx_api/models.py b/netx_api/models.py index aa58f40..a3134c1 100644 --- a/netx_api/models.py +++ b/netx_api/models.py @@ -641,6 +641,8 @@ class AppUser(Base): username: Mapped[str] = mapped_column(String(128), unique=True, index=True) password_hash: Mapped[str] = mapped_column(String(255), default="") role: Mapped[str] = mapped_column(String(32), default="user", index=True) # admin | user + # Optional capability override; empty => role defaults (see auth_scopes). + scopes: Mapped[list] = mapped_column(_JsonType, default=list) is_active: Mapped[bool] = mapped_column(Boolean, default=True, index=True) must_change_password: Mapped[bool] = mapped_column(Boolean, default=False) created_by: Mapped[str] = mapped_column(String(64), default="") @@ -675,6 +677,8 @@ class ApiToken(Base): name: Mapped[str] = mapped_column(String(128), default="") token_hash: Mapped[str] = mapped_column(String(128), unique=True, index=True) user_id: Mapped[str] = mapped_column(String(64), index=True) + # Capability subset; empty inherits owner user scopes (then intersected). + scopes: Mapped[list] = mapped_column(_JsonType, default=list) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) expires_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True, index=True) last_used_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) diff --git a/netx_api/ne_collect_runner.py b/netx_api/ne_collect_runner.py index 25129e9..94d05f9 100644 --- a/netx_api/ne_collect_runner.py +++ b/netx_api/ne_collect_runner.py @@ -45,8 +45,13 @@ def _safe_filename_part(text: str) -> str: return s[:80] or "device" -def _collect_on_device(creds: dict[str, Any], commands: list[str]) -> str: - per_cmd = int(settings.ne_collect_read_timeout_sec or 120) +def _collect_on_device( + creds: dict[str, Any], + commands: list[str], + *, + read_timeout_sec: int | None = None, +) -> str: + per_cmd = int(read_timeout_sec if read_timeout_sec is not None else (settings.ne_collect_read_timeout_sec or 120)) session_timeout = per_cmd * max(1, len(commands)) + 60 conn = open_netmiko_connection(creds, session_timeout=session_timeout) try: @@ -63,12 +68,17 @@ def _collect_on_device(creds: dict[str, Any], commands: list[str]) -> str: close_netmiko_connection(conn) -def _collect_with_timeout(creds: dict[str, Any], commands: list[str]) -> str: - per_cmd = int(settings.ne_collect_read_timeout_sec or 120) +def _collect_with_timeout( + creds: dict[str, Any], + commands: list[str], + *, + read_timeout_sec: int | None = None, +) -> str: + per_cmd = int(read_timeout_sec if read_timeout_sec is not None else (settings.ne_collect_read_timeout_sec or 120)) cap = int(settings.ne_collect_run_timeout_cap_sec or 600) budget = min(cap, per_cmd * max(1, len(commands)) + 90) with ThreadPoolExecutor(max_workers=1) as pool: - fut = pool.submit(_collect_on_device, creds, commands) + fut = pool.submit(_collect_on_device, creds, commands, read_timeout_sec=per_cmd) try: return fut.result(timeout=budget) except FuturesTimeout as exc: diff --git a/netx_api/ne_exec.py b/netx_api/ne_exec.py index eff5388..8be106f 100644 --- a/netx_api/ne_exec.py +++ b/netx_api/ne_exec.py @@ -110,10 +110,8 @@ def execute_managed_ne_commands( creds, device = resolve_cli_target(db, managed_ne_id=mid or None, ume_ne_id=uid or None) read_timeout = _normalize_read_timeout(read_timeout_sec) - prev_collect_timeout = int(settings.ne_collect_read_timeout_sec or 120) try: - settings.ne_collect_read_timeout_sec = read_timeout - output = _collect_on_device(creds, cmds) + output = _collect_on_device(creds, cmds, read_timeout_sec=read_timeout) except Exception as exc: return { "ok": False, @@ -123,8 +121,6 @@ def execute_managed_ne_commands( "error": type(exc).__name__, "detail": str(exc)[:2000], } - finally: - settings.ne_collect_read_timeout_sec = prev_collect_timeout if len(output) > _EXEC_MAX_OUTPUT: output = output[:_EXEC_MAX_OUTPUT] + "\n...[truncated]" diff --git a/netx_api/security_bootstrap.py b/netx_api/security_bootstrap.py new file mode 100644 index 0000000..03c3c70 --- /dev/null +++ b/netx_api/security_bootstrap.py @@ -0,0 +1,50 @@ +"""Startup security checks for bind address vs insecure lab defaults.""" + +from __future__ import annotations + +import logging +import sys + +from .auth_tokens import ensure_auth_secret, is_legacy_insecure_secret +from .config import settings + +_log = logging.getLogger("netx.security") + + +def _is_loopback_host(host: str) -> bool: + h = str(host or "").strip().lower() + return h in {"127.0.0.1", "::1", "localhost"} + + +def assert_secure_defaults_or_exit() -> None: + """Refuse non-loopback bind when lab secrets / default admin password remain.""" + # Always materialize per-install JWT secret before bind checks. + ensure_auth_secret() + + if bool(getattr(settings, "allow_insecure_defaults", False)): + _log.warning("NETX_ALLOW_INSECURE_DEFAULTS=1 — skipping insecure-default bind check") + return + host = str(settings.host or "127.0.0.1") + if _is_loopback_host(host): + return + problems: list[str] = [] + explicit = str(settings.auth_secret or "").strip() + if is_legacy_insecure_secret(explicit): + problems.append( + "NETX_AUTH_SECRET is still the legacy shared development value " + "(unset it to use data/auth/jwt_secret, or set a unique secret)" + ) + pwd = str(settings.bootstrap_admin_password or "").strip() + if pwd in {"", "admin123"}: + problems.append("NETX_BOOTSTRAP_ADMIN_PASSWORD is still the lab default (admin123)") + if not bool(getattr(settings, "ume_verify_tls", True)): + problems.append("NETX_UME_VERIFY_TLS=false while binding on a non-loopback interface") + if problems: + for p in problems: + _log.error("insecure default: %s", p) + _log.error( + "Refusing to start on host=%s. Set unique secrets, or bind 127.0.0.1, " + "or set NETX_ALLOW_INSECURE_DEFAULTS=1 for lab only.", + host, + ) + sys.exit(2) diff --git a/netx_api/sql_guard.py b/netx_api/sql_guard.py new file mode 100644 index 0000000..13e193b --- /dev/null +++ b/netx_api/sql_guard.py @@ -0,0 +1,138 @@ +"""Hardened read-only SQL helpers for AI / power-user query endpoints.""" + +from __future__ import annotations + +import re +from datetime import datetime, timezone +from typing import Any + +from fastapi import HTTPException +from sqlalchemy import create_engine, text as sql_text +from sqlalchemy.engine import Engine +from sqlalchemy.orm import Session, sessionmaker + +from .config import settings + +_SQL_FORBIDDEN_RE = re.compile( + r"\b(insert|update|delete|drop|alter|create|truncate|grant|revoke|call|copy|vacuum|analyze|" + r"execute|prepare|deallocate|listen|notify|load|reindex|cluster|refresh|security|" + r"set\s+role|set\s+session|into\s+outfile|pg_read_file|lo_import|lo_export)\b", + flags=re.IGNORECASE, +) +_WITH_RE = re.compile(r"^\s*with\b", flags=re.IGNORECASE) +_COMMENT_RE = re.compile(r"/\*.*?\*/|--.*?$", flags=re.IGNORECASE | re.DOTALL | re.MULTILINE) +_FROM_JOIN_RE = re.compile(r"\b(?:from|join)\s+([a-zA-Z0-9_\"\.]+)", flags=re.IGNORECASE) + +_readonly_engine: Engine | None = None +_ReadonlySession: sessionmaker | None = None + + +def _ensure_utc(value: datetime) -> datetime | None: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def _strip_sql_comments(sql: str) -> str: + return _COMMENT_RE.sub(" ", sql) + + +def validate_select_sql(sql: str, *, allowed_tables: set[str] | None = None) -> str: + cleaned = _strip_sql_comments(str(sql or "")).strip() + if not cleaned: + raise HTTPException(status_code=400, detail="sql_required") + if ";" in cleaned: + raise HTTPException(status_code=400, detail="single_statement_only") + if _WITH_RE.search(cleaned): + raise HTTPException(status_code=400, detail="with_cte_not_allowed") + low = cleaned.lower().lstrip() + if not low.startswith("select"): + raise HTTPException(status_code=400, detail="select_only") + if _SQL_FORBIDDEN_RE.search(cleaned): + raise HTTPException(status_code=400, detail="forbidden_keyword") + # Block obvious catalog / other-schema probes in the text. + if re.search(r"\bpg_catalog\b|\binformation_schema\b|\bpg_toast\b", cleaned, re.I): + raise HTTPException(status_code=400, detail="catalog_not_allowed") + if allowed_tables is not None: + refs = _FROM_JOIN_RE.findall(cleaned) + if not refs: + raise HTTPException(status_code=400, detail="from_required") + for ref in refs: + normalized = str(ref).strip().strip('"') + if "." in normalized: + normalized = normalized.split(".")[-1] + if normalized.lower() not in allowed_tables: + raise HTTPException(status_code=400, detail=f"ume_table_not_allowed:{normalized}") + return cleaned + + +def get_sql_session(db: Session) -> Session: + """Prefer dedicated read-only engine when configured; else reuse request session.""" + url = str(getattr(settings, "sql_readonly_database_url", "") or "").strip() + if not url: + return db + global _readonly_engine, _ReadonlySession + if _readonly_engine is None: + _readonly_engine = create_engine(url, pool_pre_ping=True) + _ReadonlySession = sessionmaker(bind=_readonly_engine, autoflush=False, autocommit=False) + assert _ReadonlySession is not None + return _ReadonlySession() + + +def run_select( + db: Session, + sql: str, + *, + params: dict[str, Any] | None = None, + limit: int = 200, + statement_timeout_ms: int = 0, + allowed_tables: set[str] | None = None, + require_batch_id_param: bool = False, +) -> dict[str, Any]: + limit = max(1, min(int(limit or 200), 2000)) + cleaned = validate_select_sql(sql, allowed_tables=allowed_tables) + bind_params = dict(params or {}) + if require_batch_id_param: + batch_id = str(bind_params.get("batch_id") or "").strip() + if not batch_id: + raise HTTPException(status_code=400, detail="batch_id_required") + if ":batch_id" not in cleaned: + raise HTTPException(status_code=400, detail="batch_id_param_required(:batch_id)") + wrapped = f"select * from ({cleaned}) as q limit {limit}" + own_session = False + session = db + url = str(getattr(settings, "sql_readonly_database_url", "") or "").strip() + if url: + session = get_sql_session(db) + own_session = session is not db + try: + if statement_timeout_ms > 0: + try: + if str(getattr(getattr(session, "bind", None), "dialect", None).name).lower().startswith("postgres"): + session.execute( + sql_text("SET LOCAL statement_timeout = :ms"), + {"ms": int(statement_timeout_ms)}, + ) + session.execute(sql_text("SET LOCAL search_path TO public")) + except Exception: + pass + res = session.execute(sql_text(wrapped), bind_params) + cols = list(res.keys()) + raw_rows = res.fetchall() + rows: list[list[Any]] = [] + for r in raw_rows: + out_row: list[Any] = [] + for v in list(r): + if isinstance(v, datetime): + out_row.append(((_ensure_utc(v) or v).isoformat().replace("+00:00", "Z"))) + else: + out_row.append(v) + rows.append(out_row) + return {"ok": True, "columns": cols, "rows": rows, "limit": limit} + except HTTPException: + raise + except Exception as exc: + raise HTTPException(status_code=400, detail=f"sql_failed:{str(exc)[:240]}") from exc + finally: + if own_session: + session.close() diff --git a/netx_api/sql_router.py b/netx_api/sql_router.py new file mode 100644 index 0000000..7667e8f --- /dev/null +++ b/netx_api/sql_router.py @@ -0,0 +1,47 @@ +"""Read-only SQL query routes (hardened).""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends +from sqlalchemy.orm import Session + +from .db import get_db +from .sql_guard import run_select + +router = APIRouter(tags=["sql"]) + + +@router.post("/v1/sql/query") +def sql_query(payload: dict[str, Any] | None = None, db: Session = Depends(get_db)) -> dict: + """ + Read-only SQL for legacy Excel batches. + + Safety: SELECT only, no CTE/WITH, no multi-statement, optional batch_id bind. + """ + payload = payload or {} + return run_select( + db, + str(payload.get("sql") or ""), + params={"batch_id": str(payload.get("batch_id") or "").strip()}, + limit=int(payload.get("limit") or 200), + require_batch_id_param=True, + ) + + +@router.post("/v1/sql/ume_query") +def sql_ume_query(payload: dict[str, Any] | None = None, db: Session = Depends(get_db)) -> dict: + """ + Read-only SQL for UME current alarms / inventory tables only. + """ + payload = payload or {} + timeout = int(payload.get("statement_timeout_ms") or 0) + timeout = max(0, min(timeout, 30000)) + return run_select( + db, + str(payload.get("sql") or ""), + limit=int(payload.get("limit") or 200), + statement_timeout_ms=timeout, + allowed_tables={"ume_alarms_current", "ume_inventory_ne"}, + ) diff --git a/netx_api/ume_runtime.py b/netx_api/ume_runtime.py new file mode 100644 index 0000000..c6e9745 --- /dev/null +++ b/netx_api/ume_runtime.py @@ -0,0 +1,24 @@ +"""UME / long-task runtime helpers shared by API and optional worker process. + +The API process still owns UME token keepalive, alarm WSS, and oclaw forwarder. +Config-sync / LLDP / port-traffic tick loops can run inline (default) or via +``python -m netx_api.worker`` when ``NETX_RUN_INLINE_SCHEDULERS=false``. +""" + +from __future__ import annotations + +import logging + +_log = logging.getLogger("netx.ume.runtime") + + +def start_device_schedulers() -> None: + """Start device-facing periodic collectors (safe to call once per process).""" + from .config_sync_scheduler import start_config_sync_scheduler + from .lldp_collect_scheduler import start_lldp_collect_scheduler + from .port_traffic_scheduler import start_port_traffic_scheduler + + start_config_sync_scheduler() + start_lldp_collect_scheduler() + start_port_traffic_scheduler() + _log.info("device schedulers started") diff --git a/netx_api/webcrt_router.py b/netx_api/webcrt_router.py index 6c56020..6715fad 100644 --- a/netx_api/webcrt_router.py +++ b/netx_api/webcrt_router.py @@ -13,8 +13,10 @@ from pydantic import BaseModel, Field from sqlalchemy.orm import Session from .db import SessionLocal, get_db -from .auth_deps import resolve_user_from_token +from .auth_deps import AuthContext, require_user, resolve_user_from_token +from .auth_scopes import SCOPE_WEBCRT, has_scope from .config import settings +from .webcrt_tickets import consume_ws_ticket, issue_ws_ticket from .webcrt_service import ( close_session, create_session, @@ -34,6 +36,18 @@ _log = logging.getLogger("netx.webcrt.router") router = APIRouter(prefix="/v1/webcrt", tags=["webcrt"]) +@router.post("/ws-ticket") +def api_ws_ticket(ctx: AuthContext = Depends(require_user)) -> dict[str, Any]: + """Mint a one-time short-lived ticket for WebSocket connect (Authorization header required).""" + if bool(settings.auth_enabled) and not has_scope(ctx.scopes, SCOPE_WEBCRT): + raise HTTPException( + status_code=403, + detail={"error": "insufficient_scope", "required": [SCOPE_WEBCRT], "granted": sorted(ctx.scopes)}, + ) + ticket, ttl = issue_ws_ticket(user_id=str(ctx.user.id), scopes=ctx.scopes) + return {"ticket": ticket, "expires_in": ttl} + + class WebcrtSessionCreate(BaseModel): ne_id: str | None = Field(default=None) ume_ne_id: str | None = Field(default=None) @@ -342,19 +356,34 @@ async def api_sftp_upload( @router.websocket("/sessions/{session_id}/ws") async def websocket_session(websocket: WebSocket, session_id: str) -> None: if bool(settings.auth_enabled): - token = str(websocket.query_params.get("access_token") or "").strip() - if not token: + # Prefer one-time ws_ticket (query). Bearer header also accepted (non-browser clients). + # Long-lived access_token in query is rejected. + ticket = str(websocket.query_params.get("ws_ticket") or "").strip() + if ticket: + info = consume_ws_ticket(ticket) + if info is None or not has_scope(info.scopes, SCOPE_WEBCRT): + await websocket.close(code=4403 if info is not None else 4401) + return + else: + if str(websocket.query_params.get("access_token") or "").strip(): + await websocket.close(code=4401) + return + token = "" auth = str(websocket.headers.get("authorization") or "").strip() if auth.lower().startswith("bearer "): token = auth[7:].strip() - db = SessionLocal() - try: - resolved = resolve_user_from_token(db, token) if token else None - finally: - db.close() - if resolved is None: - await websocket.close(code=4401) - return + db = SessionLocal() + try: + resolved = resolve_user_from_token(db, token) if token else None + finally: + db.close() + if resolved is None: + await websocket.close(code=4401) + return + _user, _via, scopes, _tid = resolved + if not has_scope(scopes, SCOPE_WEBCRT): + await websocket.close(code=4403) + return await websocket.accept() attach_gen = 0 diff --git a/netx_api/webcrt_tickets.py b/netx_api/webcrt_tickets.py new file mode 100644 index 0000000..241fa72 --- /dev/null +++ b/netx_api/webcrt_tickets.py @@ -0,0 +1,49 @@ +"""Short-lived WebCRT WebSocket tickets (avoid putting JWT in query strings).""" + +from __future__ import annotations + +import secrets +import threading +import time +from dataclasses import dataclass + +_TTL_SEC = 90 +_lock = threading.Lock() +_tickets: dict[str, tuple[float, str, frozenset[str]]] = {} + + +@dataclass(frozen=True) +class TicketInfo: + user_id: str + scopes: frozenset[str] + + +def issue_ws_ticket(*, user_id: str, scopes: frozenset[str], ttl_sec: int = _TTL_SEC) -> tuple[str, int]: + tid = secrets.token_urlsafe(24) + exp = time.time() + max(15, int(ttl_sec)) + with _lock: + _purge_locked() + _tickets[tid] = (exp, str(user_id), frozenset(scopes)) + return tid, max(15, int(ttl_sec)) + + +def consume_ws_ticket(ticket: str) -> TicketInfo | None: + raw = str(ticket or "").strip() + if not raw: + return None + with _lock: + _purge_locked() + row = _tickets.pop(raw, None) + if row is None: + return None + exp, user_id, scopes = row + if exp < time.time(): + return None + return TicketInfo(user_id=user_id, scopes=scopes) + + +def _purge_locked() -> None: + now = time.time() + dead = [k for k, (exp, _, _) in _tickets.items() if exp < now] + for k in dead: + _tickets.pop(k, None) diff --git a/netx_api/worker.py b/netx_api/worker.py new file mode 100644 index 0000000..1bcb63d --- /dev/null +++ b/netx_api/worker.py @@ -0,0 +1,49 @@ +"""Background worker process for long-running schedulers. + +Run separately from the API when NETX_RUN_INLINE_SCHEDULERS=false: + + python -m netx_api.worker + +Starts: config_sync, lldp_collect, port_traffic tick loops. +UME WS / keepalive remain in the API process (token + alarm coordination). +""" + +from __future__ import annotations + +import logging +import signal +import threading +import time + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", +) +_log = logging.getLogger("netx.worker") + + +def main() -> None: + stop = threading.Event() + + def _handle(_sig: int, _frame: object) -> None: + _log.info("shutdown signal received") + stop.set() + + for sig in (signal.SIGINT, signal.SIGTERM): + try: + signal.signal(sig, _handle) + except Exception: + pass + + from .ume_runtime import start_device_schedulers + + start_device_schedulers() + _log.info("netx worker schedulers started (config_sync, lldp_collect, port_traffic)") + + while not stop.is_set(): + time.sleep(1.0) + _log.info("netx worker exiting") + + +if __name__ == "__main__": + main() diff --git a/packages/netx-mcp/src/netx_mcp/http_tools.py b/packages/netx-mcp/src/netx_mcp/http_tools.py index 83f8602..ed1d62a 100644 --- a/packages/netx-mcp/src/netx_mcp/http_tools.py +++ b/packages/netx-mcp/src/netx_mcp/http_tools.py @@ -571,6 +571,40 @@ _HANDLERS: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = { "queryTopologyEdges": _query_topology_edges, } +# Minimum scope required to advertise / invoke each tool (matches netx API RBAC). +TOOL_REQUIRED_SCOPE: dict[str, str] = { + "queryUmeAlarms": "alarms:read", + "aggregateUmeAlarms": "alarms:read", + "runUmeDiagnostics": "alarms:read", + "queryUmeNeInventory": "ne:read", + "getUmeNe": "ne:read", + "queryUmeAlarmsRaw": "alarms:read", + "aggregateUmeAlarmsRaw": "alarms:read", + "listUmeAlarmFields": "alarms:read", + "sqlQueryUme": "sql:query", + "listManagedNe": "ne:read", + "getManagedNe": "ne:read", + "execManagedNe": "ne:exec", + "listCliTargets": "ne:read", + "queryTopologyEdges": "ne:read", +} + + +def tools_for_scopes(scopes: list[str] | set[str] | frozenset[str] | None) -> list[dict[str, Any]]: + """Filter MCP tool list by granted scopes. Empty/None => return all (offline / unauthenticated listing).""" + if scopes is None: + return list(HTTP_MCP_TOOLS) + granted = {str(s).strip().lower() for s in scopes if str(s).strip()} + if not granted: + return [] + out: list[dict[str, Any]] = [] + for tool in HTTP_MCP_TOOLS: + name = str(tool.get("name") or "") + need = TOOL_REQUIRED_SCOPE.get(name) + if need is None or need in granted: + out.append(tool) + return out + def call_http_tool(name: str, args: dict[str, Any]) -> dict[str, Any]: fn = _HANDLERS.get(str(name or "").strip()) diff --git a/packages/netx-mcp/src/netx_mcp/server.py b/packages/netx-mcp/src/netx_mcp/server.py index 82d13d8..2448267 100644 --- a/packages/netx-mcp/src/netx_mcp/server.py +++ b/packages/netx-mcp/src/netx_mcp/server.py @@ -12,7 +12,8 @@ import json import sys from typing import Any -from netx_mcp.http_tools import HTTP_MCP_TOOLS, call_http_tool +from netx_mcp.http_client import http_json +from netx_mcp.http_tools import TOOL_REQUIRED_SCOPE, call_http_tool, tools_for_scopes def _ensure_utf8_stdio() -> None: @@ -39,7 +40,30 @@ def _err(rid: Any, code: int, message: str) -> None: sys.stdout.flush() +def _fetch_scopes() -> list[str] | None: + """Return granted scopes from /v1/auth/me, or None if the call fails (show all tools).""" + try: + data = http_json("GET", "/v1/auth/me") + scopes = data.get("scopes") if isinstance(data, dict) else None + if isinstance(scopes, list): + return [str(s) for s in scopes] + user = data.get("user") if isinstance(data, dict) else None + if isinstance(user, dict) and isinstance(user.get("scopes"), list): + return [str(s) for s in user["scopes"]] + except Exception: + return None + return None + + def run_stdio_loop() -> None: + cached_scopes: list[str] | None | object = object() + + def scopes() -> list[str] | None: + nonlocal cached_scopes + if cached_scopes is object(): + cached_scopes = _fetch_scopes() + return cached_scopes # type: ignore[return-value] + for line in sys.stdin: raw = line.strip() if not raw: @@ -59,17 +83,22 @@ def run_stdio_loop() -> None: { "protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, - "serverInfo": {"name": "netx-mcp", "version": "0.2.0", "mode": "http"}, + "serverInfo": {"name": "netx-mcp", "version": "0.2.1", "mode": "http"}, }, ) continue if method == "notifications/initialized": continue if method == "tools/list": - _ok(rid, {"tools": HTTP_MCP_TOOLS}) + _ok(rid, {"tools": tools_for_scopes(scopes())}) continue if method == "tools/call": name = str(params.get("name") or "") + need = TOOL_REQUIRED_SCOPE.get(name) + granted = scopes() + if need and granted is not None and need not in {str(s).lower() for s in granted}: + _err(rid, -32001, f"insufficient_scope:{need}") + continue args = params.get("arguments") if isinstance(params.get("arguments"), dict) else {} _ok(rid, call_http_tool(name, args)) continue diff --git a/pyproject.toml b/pyproject.toml index e9dba5c..150cecd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,9 @@ dependencies = [ "websocket-client>=1.8.0", "cryptography>=42.0.0", "netmiko>=4.3.0", + "bcrypt>=4.1.0", + "PyJWT>=2.8.0", + "alembic>=1.13.0", ] [project.optional-dependencies] diff --git a/requirements.txt b/requirements.txt index 1be1699..602552b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -15,3 +15,4 @@ cryptography>=42.0.0 netmiko>=4.3.0 bcrypt>=4.1.0 PyJWT>=2.8.0 +alembic>=1.13.0 diff --git a/tests/test_auth.py b/tests/test_auth.py index 96df175..96722fe 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -68,6 +68,7 @@ class AuthApiTests(unittest.TestCase): self._sess_patch.start() self._settings_patches = [ patch("netx_api.auth_middleware.settings.auth_enabled", True), + patch("netx_api.audit_async.settings.audit_async", False), patch("netx_api.auth_tokens.settings.auth_secret", "unit-test-auth-secret-32bytes!!"), patch("netx_api.auth_tokens.settings.auth_token_ttl_sec", 3600), patch("netx_api.auth_service.settings.bootstrap_admin_username", "admin"), diff --git a/tests/test_rbac_scopes.py b/tests/test_rbac_scopes.py new file mode 100644 index 0000000..d20ed79 --- /dev/null +++ b/tests/test_rbac_scopes.py @@ -0,0 +1,237 @@ +"""RBAC scopes, SQL guard, and insecure-default startup checks.""" + +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from fastapi import FastAPI, HTTPException +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + +from netx_api.auth_middleware import AuthAuditMiddleware +from netx_api.auth_router import router as auth_router +from netx_api.auth_scopes import ( + MCP_DEFAULT_SCOPES, + SCOPE_SQL, + SCOPE_WEBCRT, + effective_token_scopes, + required_scope_for_request, + scopes_for_role, +) +from netx_api.auth_service import bootstrap_admin_if_needed, create_api_token, create_user +from netx_api.db import Base, get_db +from netx_api.models import AppUser +from netx_api.security_bootstrap import assert_secure_defaults_or_exit +from netx_api.sql_guard import validate_select_sql +from netx_api.sql_router import router as sql_router + + +class ScopeUnitTests(unittest.TestCase): + def test_role_defaults(self) -> None: + self.assertIn(SCOPE_WEBCRT, scopes_for_role("admin")) + self.assertNotIn(SCOPE_WEBCRT, scopes_for_role("user")) + self.assertNotIn(SCOPE_SQL, scopes_for_role("user")) + self.assertIn("alarms:read", scopes_for_role("user")) + + def test_mcp_default_excludes_webcrt_sql(self) -> None: + self.assertNotIn(SCOPE_WEBCRT, MCP_DEFAULT_SCOPES) + self.assertNotIn(SCOPE_SQL, MCP_DEFAULT_SCOPES) + self.assertIn("ne:exec", MCP_DEFAULT_SCOPES) + + def test_token_intersection(self) -> None: + user = {"alarms:read", "ne:read", "ne:exec", "sql:query"} + tok = effective_token_scopes(user_scopes=user, token_scopes=["ne:exec", "sql:query", "webcrt:session"]) + self.assertEqual(tok, frozenset({"ne:exec", "sql:query"})) + + def test_path_scope_map(self) -> None: + self.assertEqual(required_scope_for_request("POST", "/v1/sql/ume_query"), SCOPE_SQL) + self.assertEqual(required_scope_for_request("GET", "/v1/webcrt/sessions"), SCOPE_WEBCRT) + self.assertEqual(required_scope_for_request("POST", "/v1/managed-ne/exec"), "ne:exec") + self.assertEqual(required_scope_for_request("POST", "/v1/managed-ne"), "ne:write") + + +class SqlGuardTests(unittest.TestCase): + def test_rejects_cte(self) -> None: + with self.assertRaises(HTTPException) as ctx: + validate_select_sql( + "WITH x AS (SELECT * FROM app_user) SELECT * FROM x", + allowed_tables={"ume_alarms_current", "ume_inventory_ne"}, + ) + self.assertEqual(ctx.exception.detail, "with_cte_not_allowed") + + def test_rejects_disallowed_table(self) -> None: + with self.assertRaises(HTTPException) as ctx: + validate_select_sql( + "select * from app_user", + allowed_tables={"ume_alarms_current", "ume_inventory_ne"}, + ) + self.assertIn("ume_table_not_allowed", str(ctx.exception.detail)) + + def test_rejects_catalog(self) -> None: + with self.assertRaises(HTTPException) as ctx: + validate_select_sql( + "select * from ume_alarms_current where ne_id in (select relname from pg_catalog.pg_class)", + allowed_tables={"ume_alarms_current", "ume_inventory_ne"}, + ) + detail = str(ctx.exception.detail) + self.assertTrue( + detail == "catalog_not_allowed" or detail.startswith("ume_table_not_allowed:"), + detail, + ) + + +class SecurityBootstrapTests(unittest.TestCase): + def test_loopback_allows_defaults(self) -> None: + with patch("netx_api.security_bootstrap.settings") as st: + st.allow_insecure_defaults = False + st.host = "127.0.0.1" + st.auth_secret = "" + st.bootstrap_admin_password = "admin123" + st.ume_verify_tls = False + with patch("netx_api.security_bootstrap.ensure_auth_secret", return_value="x" * 48): + assert_secure_defaults_or_exit() + + def test_non_loopback_rejects_legacy_secret(self) -> None: + with patch("netx_api.security_bootstrap.settings") as st: + st.allow_insecure_defaults = False + st.host = "0.0.0.0" + st.auth_secret = "netx-dev-auth-secret-change-me-in-production-32b" + st.bootstrap_admin_password = "strong-pass-here" + st.ume_verify_tls = True + with patch("netx_api.security_bootstrap.ensure_auth_secret", return_value="x" * 48): + with self.assertRaises(SystemExit) as ctx: + assert_secure_defaults_or_exit() + self.assertEqual(ctx.exception.code, 2) + + def test_non_loopback_rejects_default_password(self) -> None: + with patch("netx_api.security_bootstrap.settings") as st: + st.allow_insecure_defaults = False + st.host = "0.0.0.0" + st.auth_secret = "" + st.bootstrap_admin_password = "admin123" + st.ume_verify_tls = True + with patch("netx_api.security_bootstrap.ensure_auth_secret", return_value="x" * 48): + with self.assertRaises(SystemExit) as ctx: + assert_secure_defaults_or_exit() + self.assertEqual(ctx.exception.code, 2) + + +class RbacApiTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine( + "sqlite+pysqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + Base.metadata.create_all(bind=self.engine) + self.Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False) + + self.app = FastAPI() + self.app.add_middleware(AuthAuditMiddleware) + self.app.include_router(auth_router) + self.app.include_router(sql_router) + + @self.app.post("/v1/webcrt/sessions") + def fake_webcrt() -> dict[str, str]: + return {"ok": "1"} + + @self.app.post("/v1/managed-ne/exec") + def fake_exec() -> dict[str, str]: + return {"ok": "1"} + + def _override_db(): + db = self.Session() + try: + yield db + finally: + db.close() + + self.app.dependency_overrides[get_db] = _override_db + self._patches = [ + patch("netx_api.auth_middleware.SessionLocal", self.Session), + patch("netx_api.audit_async.settings.audit_async", False), + patch("netx_api.auth_middleware.settings.auth_enabled", True), + patch("netx_api.auth_tokens.settings.auth_secret", "unit-test-auth-secret-32bytes!!"), + patch("netx_api.auth_tokens.settings.auth_token_ttl_sec", 3600), + patch("netx_api.auth_service.settings.bootstrap_admin_username", "admin"), + patch("netx_api.auth_service.settings.bootstrap_admin_password", "adminpass"), + patch("netx_api.auth_deps.settings.auth_enabled", True), + ] + for p in self._patches: + p.start() + + db = self.Session() + try: + bootstrap_admin_if_needed(db) + admin = db.query(AppUser).filter(AppUser.username == "admin").one() + create_user(db, username="alice", password="alice12", role="user", actor=admin) + finally: + db.close() + self.client = TestClient(self.app) + + def tearDown(self) -> None: + for p in self._patches: + p.stop() + self.app.dependency_overrides.clear() + self.engine.dispose() + + def _login(self, username: str, password: str) -> str: + r = self.client.post("/v1/auth/login", json={"username": username, "password": password}) + self.assertEqual(r.status_code, 200, r.text) + return str(r.json()["access_token"]) + + def test_user_denied_webcrt_and_sql(self) -> None: + token = self._login("alice", "alice12") + h = {"Authorization": f"Bearer {token}"} + self.assertEqual(self.client.post("/v1/webcrt/sessions", headers=h).status_code, 403) + self.assertEqual( + self.client.post( + "/v1/sql/ume_query", + headers=h, + json={"sql": "select * from ume_alarms_current", "limit": 1}, + ).status_code, + 403, + ) + self.assertEqual(self.client.post("/v1/managed-ne/exec", headers=h, json={}).status_code, 403) + + def test_admin_allowed_sql_path_auth(self) -> None: + token = self._login("admin", "adminpass") + h = {"Authorization": f"Bearer {token}"} + # Passes scope gate; may fail SQL execution without tables — not 403. + r = self.client.post( + "/v1/sql/ume_query", + headers=h, + json={"sql": "select 1 as n from ume_alarms_current", "limit": 1}, + ) + self.assertNotEqual(r.status_code, 403) + + def test_me_returns_scopes(self) -> None: + token = self._login("alice", "alice12") + me = self.client.get("/v1/auth/me", headers={"Authorization": f"Bearer {token}"}) + self.assertEqual(me.status_code, 200) + scopes = me.json()["scopes"] + self.assertIn("alarms:read", scopes) + self.assertNotIn("webcrt:session", scopes) + + def test_mcp_token_scopes(self) -> None: + token = self._login("admin", "adminpass") + created = self.client.post( + "/v1/api-tokens", + headers={"Authorization": f"Bearer {token}"}, + json={"name": "mcp", "expires_in_days": 0, "scopes": list(MCP_DEFAULT_SCOPES)}, + ) + self.assertEqual(created.status_code, 200, created.text) + plain = created.json()["token"]["token"] + # MCP token cannot call webcrt + r = self.client.post( + "/v1/webcrt/sessions", + headers={"Authorization": f"Bearer {plain}"}, + ) + self.assertEqual(r.status_code, 403) + + +if __name__ == "__main__": + unittest.main() diff --git a/web/src/auth/AuthContext.tsx b/web/src/auth/AuthContext.tsx index 2c6c896..e99dca7 100644 --- a/web/src/auth/AuthContext.tsx +++ b/web/src/auth/AuthContext.tsx @@ -20,6 +20,7 @@ export type AuthUser = { id: string; username: string; role: string; + scopes?: string[]; is_active: boolean; must_change_password?: boolean; created_by?: string; @@ -31,10 +32,12 @@ type AuthState = { ready: boolean; token: string | null; user: AuthUser | null; + scopes: string[]; login: (username: string, password: string) => Promise; logout: () => Promise; refreshMe: () => Promise; isAdmin: boolean; + hasScope: (scope: string) => boolean; }; const AuthContext = createContext(null); @@ -43,22 +46,26 @@ export function AuthProvider({ children }: { children: ReactNode }) { const [ready, setReady] = useState(false); const [token, setToken] = useState(() => getAuthToken()); const [user, setUser] = useState(null); + const [scopes, setScopes] = useState([]); const refreshMe = useCallback(async () => { const tok = getAuthToken(); if (!tok) { setToken(null); setUser(null); + setScopes([]); return; } try { - const data = await apiGet<{ user: AuthUser }>("/v1/auth/me"); + const data = await apiGet<{ user: AuthUser; scopes?: string[] }>("/v1/auth/me"); setToken(tok); setUser(data.user); + setScopes(data.scopes || data.user.scopes || []); } catch { clearAuthToken(); setToken(null); setUser(null); + setScopes([]); } }, []); @@ -78,6 +85,7 @@ export function AuthProvider({ children }: { children: ReactNode }) { if (ev.key === null || ev.newValue == null || ev.newValue === "") { setToken(null); setUser(null); + setScopes([]); return; } void refreshMe(); @@ -94,6 +102,7 @@ export function AuthProvider({ children }: { children: ReactNode }) { setAuthToken(data.access_token); setToken(data.access_token); setUser(data.user); + setScopes(data.user.scopes || []); }, []); const logout = useCallback(async () => { @@ -107,6 +116,7 @@ export function AuthProvider({ children }: { children: ReactNode }) { clearAuthToken(); setToken(null); setUser(null); + setScopes([]); }, []); const value = useMemo( @@ -114,12 +124,14 @@ export function AuthProvider({ children }: { children: ReactNode }) { ready, token, user, + scopes, login, logout, refreshMe, isAdmin: user?.role === "admin", + hasScope: (scope: string) => scopes.includes(scope) || user?.role === "admin", }), - [ready, token, user, login, logout, refreshMe], + [ready, token, user, scopes, login, logout, refreshMe], ); return {children}; diff --git a/web/src/config/modules.ts b/web/src/config/modules.ts index 4cda705..83254e0 100644 --- a/web/src/config/modules.ts +++ b/web/src/config/modules.ts @@ -13,6 +13,8 @@ export type ModuleDefinition = { descKey?: string; iconTone: ModuleIconTone; titleKey: string; + /** Required capability scope to show in workbench (admin bypasses). */ + requiredScope?: string; adminOnly?: boolean; /** Hide from workbench launcher; still used for module window registration. */ workbenchHidden?: boolean; @@ -63,6 +65,7 @@ export const MODULES: readonly ModuleDefinition[] = [ descKey: "workbench.cards.webcrtDesc", iconTone: "slate", titleKey: "layout.titleWebcrt", + requiredScope: "webcrt:session", }, { moduleId: "port-traffic-wall", diff --git a/web/src/pages/WebcrtPage.tsx b/web/src/pages/WebcrtPage.tsx index b978e61..06bde17 100644 --- a/web/src/pages/WebcrtPage.tsx +++ b/web/src/pages/WebcrtPage.tsx @@ -817,7 +817,7 @@ export function WebcrtPage() { const pending: TermTab = { key, sessionId, - wsUrl: webcrtWsUrl(sessionId), + wsUrl: "", termEpoch: (existing?.termEpoch || 0) + 1, target, status: "connecting", @@ -833,9 +833,12 @@ export function WebcrtPage() { return [...without, pending]; }); setActiveTabKey(key); + void webcrtWsUrl(sessionId).then((wsUrl) => { + updateTab(key, { wsUrl }); + }); showOk(t("webcrt.opened", { name: deviceLabel(target) })); }, - [showOk, t], + [showOk, t, updateTab], ); const openAuthForTarget = useCallback((target: CliTargetItem, errorHint?: string) => { @@ -927,7 +930,7 @@ export function WebcrtPage() { async_connect: true, }; const sess = await createWebcrtSession(body); - const wsUrl = webcrtWsUrl(sess.session_id); + const wsUrl = await webcrtWsUrl(sess.session_id); updateTab(key, { sessionId: sess.session_id, wsUrl, @@ -966,9 +969,12 @@ export function WebcrtPage() { status: "connecting", connectPhase: "authenticating", termEpoch: tab.termEpoch + 1, - wsUrl: webcrtWsUrl(tab.sessionId), + wsUrl: "", errorMessage: undefined, }); + void webcrtWsUrl(tab.sessionId).then((wsUrl) => { + updateTab(tab.key, { wsUrl }); + }); setActiveTabKey(tab.key); return true; }, diff --git a/web/src/pages/WorkbenchPage.tsx b/web/src/pages/WorkbenchPage.tsx index 758287c..e53c34e 100644 --- a/web/src/pages/WorkbenchPage.tsx +++ b/web/src/pages/WorkbenchPage.tsx @@ -8,7 +8,7 @@ const SECTIONS: WorkbenchSection[] = ["monitoring", "operations", "system"]; export function WorkbenchPage() { const { t } = useI18n(); - const { isAdmin } = useAuth(); + const { isAdmin, hasScope } = useAuth(); return (
@@ -19,6 +19,7 @@ export function WorkbenchPage() { {modulesInSection(section) .filter((mod) => !mod.workbenchHidden) .filter((mod) => !mod.adminOnly || isAdmin) + .filter((mod) => !mod.requiredScope || hasScope(mod.requiredScope) || isAdmin) .map((mod) => (