Merge pull request #336 from ipnet-mesh/feat/routes-ownership

feat(routes): ownership-based edit/delete permissions
This commit is contained in:
JingleManSweep
2026-07-24 22:02:03 +01:00
committed by GitHub
11 changed files with 729 additions and 38 deletions
@@ -0,0 +1,36 @@
"""add routes.created_by
Revision ID: da69304d8106
Revises: d307ee761a34
Create Date: 2026-07-24 10:00:00.000000+00:00
Adds a nullable ``created_by`` column to ``routes``, storing the OIDC
subject identifier (``user_id``) of the operator or admin who created
the route. This enables ownership-based write permissions: operators
can only modify routes they created, while admins can modify any route
and take ownership on edit.
Existing routes get ``NULL`` (admin-only modification), preserving
backwards compatibility with no data backfill.
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "da69304d8106"
down_revision: Union[str, None] = "d307ee761a34"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
with op.batch_alter_table("routes", schema=None) as batch_op:
batch_op.add_column(sa.Column("created_by", sa.String(255), nullable=True))
def downgrade() -> None:
with op.batch_alter_table("routes", schema=None) as batch_op:
batch_op.drop_column("created_by")
+1 -1
View File
@@ -21,7 +21,7 @@ User roles are read from the OIDC token's `roles` claim (configurable via `OIDC_
| Role | Config Variable | Default | Description |
|------|----------------|---------|-------------|
| Admin | `OIDC_ROLE_ADMIN` | `admin` | Full write access to all API endpoints through the proxy |
| Operator | `OIDC_ROLE_OPERATOR` | `operator` | Manage nodes, node tags, adoptions, and routes (create/edit/delete, scoped to the operator visibility tier) |
| Operator | `OIDC_ROLE_OPERATOR` | `operator` | Manage nodes, node tags, adoptions, and routes (create; edit/delete only routes they created) |
| Member | `OIDC_ROLE_MEMBER` | `member` | Read-only access (no endpoint assignments) |
The role names are configurable to match your IdP's role naming convention. For example, if your IdP uses `superuser` instead of `admin`, set `OIDC_ROLE_ADMIN=superuser`.
+4 -2
View File
@@ -44,7 +44,9 @@ The collector runs a background thread that re-evaluates every enabled route on
Routes carry the same role-based visibility levels as channels — `community`, `member`, `operator`, `admin`. A user only sees routes whose visibility is at or below their role's maximum level. Seeded routes default to `community` (visible to everyone); set a higher level to restrict a route to operators/admins only. Visibility is enforced on both the list and detail endpoints, so a hidden route's existence is not leaked.
Both operators and admins can create, edit, and delete routes. A user may never scope a route above their own role (e.g. an operator cannot create an `admin`-visibility route) — this is enforced on the write endpoints and prevents a user from creating a route they could then never see or modify. Operators can only edit/delete routes whose visibility is at or below the operator tier; attempting to modify a higher-visibility route returns `404`.
Both operators and admins can create routes. A user may never scope a route above their own role (e.g. an operator cannot create an `admin`-visibility route) — this is enforced on the write endpoints and prevents a user from creating a route they could then never see or modify.
**Ownership-based editing:** Each route records the user who created it (`created_by`). Operators can edit and delete only the routes they created. Admins can edit and delete any route; they claim ownership of legacy (unowned) routes — those with a `NULL` `created_by` — when they edit them, but do not displace an existing creator. Routes created before ownership tracking was introduced have a `NULL` `created_by` and are admin-only. The creator's display name is shown on each route card when available. Attempting to modify a route above the caller's visibility tier returns `404`; modifying a visible route the caller does not own returns `403`.
## Defining routes
@@ -53,4 +55,4 @@ Routes are keyed by their `from`/`to` endpoint labels and upserted by that pair.
- **Seed YAML** — add a `routes.yaml` to your `SEED_HOME` and run the seed process. See [seeding.md → Routes](seeding.md#routes) for the format and rules (path nodes must already exist in the database; the `(from, to)` pair must be unique).
- **API** — `POST /api/v1/routes` (operator or admin) creates a route, with a `/preview` endpoint that dry-runs matching against an unsaved configuration so you can tune thresholds before committing. See `SCHEMAS.md` for the request/response shapes.
The `/routes` page renders the live status card, the per-day history strip, recent matching transmissions (with observer attribution), and — for operators and admins — inline edit/delete controls.
The `/routes` page renders the live status card, the per-day history strip, recent matching transmissions (with observer attribution), the route owner's name, and inline edit/delete controls gated per-route by ownership (operators see controls only on their own routes; admins see them on all routes).
+9 -6
View File
@@ -12,13 +12,16 @@ test.describe.serial("routes (operator)", () => {
await page.goto("/routes");
// Operators see the seeded community route and the add button.
await expect(
page.locator(
'[data-testid="route-card"][data-route-label="Alpha Site \u2192 Bravo Site"]',
),
).toBeVisible();
const seededCard = page.locator(
'[data-testid="route-card"][data-route-label="Alpha Site \u2192 Bravo Site"]',
);
await expect(seededCard).toBeVisible();
await expect(page.getByTestId("add-route")).toBeVisible();
// The seeded route has NULL created_by — operator must NOT see edit/delete.
await expect(seededCard.getByTestId("edit-route")).toHaveCount(0);
await expect(seededCard.getByTestId("delete-route")).toHaveCount(0);
// The visibility dropdown must NOT offer the admin tier to an operator.
await page.getByTestId("add-route").click();
const modal = page.locator('[data-testid="route-modal"]');
@@ -45,7 +48,7 @@ test.describe.serial("routes (operator)", () => {
);
await expect(card).toBeVisible();
// Operator can edit the route they created.
// Operator owns the route they just created — edit/delete should be present.
await card.getByTestId("edit-route").click();
await expect(modal).toBeVisible();
await expect(page.getByTestId("route-from")).toHaveValue("Op From");
+8 -3
View File
@@ -11,9 +11,14 @@ test.describe.serial("routes (admin)", () => {
}) => {
await page.goto("/routes");
await expect(
page.locator('[data-testid="route-card"][data-route-label="Alpha Site \u2192 Bravo Site"]'),
).toBeVisible();
const seededCard = page.locator(
'[data-testid="route-card"][data-route-label="Alpha Site \u2192 Bravo Site"]',
);
await expect(seededCard).toBeVisible();
// Admin sees edit/delete on all routes, including legacy (NULL created_by) ones.
await expect(seededCard.getByTestId("edit-route")).toBeVisible();
await expect(seededCard.getByTestId("delete-route")).toBeVisible();
await page.getByTestId("add-route").click();
const modal = page.locator('[data-testid="route-modal"]');
+98 -14
View File
@@ -1,12 +1,13 @@
"""Route health monitoring API routes."""
import logging
from datetime import datetime, timedelta, timezone
from typing import Any
from fastapi import APIRouter, HTTPException, Request
from sqlalchemy import select
from meshcore_hub.api.auth import RequireOperatorOrAdmin, RequireRead
from meshcore_hub.api.auth import RequireOperatorOrAdmin, RequireRead, X_USER_ID_HEADER
from meshcore_hub.api.cache import cached, sorted_query_string
from meshcore_hub.api.cache_invalidation import invalidate_routes
from meshcore_hub.api.channel_visibility import (
@@ -15,6 +16,7 @@ from meshcore_hub.api.channel_visibility import (
resolve_user_role,
)
from meshcore_hub.api.dependencies import DbSession
from meshcore_hub.api.profile_utils import get_or_create_profile
from meshcore_hub.collector.routes import (
compute_persisted_quality_avg,
derive_expected_hash,
@@ -34,6 +36,7 @@ from meshcore_hub.common.models.route_node import RouteNode
from meshcore_hub.common.models.route_observer import RouteObserver
from meshcore_hub.common.models.route_recent_match import RouteRecentMatch
from meshcore_hub.common.models.route_result import RouteResult
from meshcore_hub.common.models.user_profile import UserProfile
from meshcore_hub.common.schemas.routes import (
ContributingObserver,
RecentMatchPath,
@@ -44,6 +47,7 @@ from meshcore_hub.common.schemas.routes import (
RouteList,
RouteNodeRead,
RouteObserverRead,
RouteOwner,
RoutePreviewRequest,
RoutePreviewResponse,
RouteRead,
@@ -52,6 +56,7 @@ from meshcore_hub.common.schemas.routes import (
)
router = APIRouter()
logger = logging.getLogger(__name__)
# Sentinel distinguishing "use the precomputed value" from "explicitly None"
# (the latter is what create_route passes to preserve the "brand-new route
@@ -88,15 +93,28 @@ def _assert_visibility_within_role(request: Request, visibility: str) -> None:
def _assert_route_modifiable(request: Request, route: Route) -> None:
"""Reject modifying a route above the caller's visibility tier.
"""Reject modifying a route the caller is not allowed to touch.
Returns 404 (mirroring the GET detail behaviour) so the existence of a
higher-visibility route is not leaked to lower-privileged callers.
Layer 1 — visibility: a route above the caller's tier yields **404**
so its existence is not leaked (mirrors GET detail behaviour).
Layer 2 — ownership: operators may only modify routes *they* created.
A visible-but-unowned route yields **403** (the caller can see it in
the list, so a transparent rejection is better UX). ``created_by``
is ``None`` for legacy routes (pre-ownership-tracking) and is treated
as admin-only. Admins bypass the ownership check entirely.
"""
if VISIBILITY_LEVELS.get(route.visibility, 0) > _caller_max_visibility_level(
request
):
raise HTTPException(status_code=404, detail="Route not found")
if resolve_user_role(request) != "admin":
caller_id = request.headers.get(X_USER_ID_HEADER, "")
if route.created_by is None or route.created_by != caller_id:
raise HTTPException(
status_code=403,
detail="You can only modify routes you created",
)
def _route_node_to_read(rn: RouteNode) -> RouteNodeRead:
@@ -130,13 +148,58 @@ def _result_to_summary(result: RouteResult | None) -> RouteResultSummary | None:
)
def _route_to_read(route: Route, *, quality_avg: Any = _UNSET) -> RouteRead:
def _profile_to_owner(profile: UserProfile) -> RouteOwner:
"""Convert a UserProfile to the lightweight RouteOwner display schema."""
return RouteOwner(
user_id=profile.user_id,
name=profile.name,
callsign=profile.callsign,
profile_id=profile.id,
)
def _resolve_owner(session: DbSession, created_by: str | None) -> UserProfile | None:
"""Resolve a single ``created_by`` user_id to a UserProfile."""
if not created_by:
return None
return session.execute(
select(UserProfile).where(UserProfile.user_id == created_by)
).scalar_one_or_none()
def _resolve_owners_batch(
session: DbSession, routes: list[Route]
) -> dict[str, UserProfile]:
"""Batch-resolve creator profiles for a list of routes (avoids N+1)."""
owner_ids = {r.created_by for r in routes if r.created_by}
if not owner_ids:
return {}
return {
p.user_id: p
for p in session.execute(
select(UserProfile).where(UserProfile.user_id.in_(owner_ids))
)
.scalars()
.all()
}
def _route_to_read(
route: Route,
*,
quality_avg: Any = _UNSET,
owner: UserProfile | None = None,
) -> RouteRead:
"""Serialize a Route to its list-level read schema.
``quality_avg`` defaults to the precomputed value persisted on
``route.route_result.quality_avg`` (written by the background
evaluator). Callers may pass an explicit value (e.g. ``None`` on
create responses) to override.
``owner`` is the resolved UserProfile for ``route.created_by``, if
any. Callers should pass it in to avoid per-row queries in list
contexts (use ``_resolve_owners_batch``).
"""
if quality_avg is _UNSET:
quality_avg = route.route_result.quality_avg if route.route_result else None
@@ -158,6 +221,8 @@ def _route_to_read(route: Route, *, quality_avg: Any = _UNSET) -> RouteRead:
route_observers=[_route_observer_to_read(ro) for ro in route.route_observers],
route_result=_result_to_summary(route.route_result),
quality_avg=quality_avg,
created_by=route.created_by,
owner=_profile_to_owner(owner) if owner else None,
created_at=route.created_at,
updated_at=route.updated_at,
)
@@ -257,22 +322,27 @@ def list_routes(
max_level = get_max_visibility_level(role)
routes = session.execute(select(Route).order_by(Route.from_label)).scalars().all()
visible = [r for r in routes if VISIBILITY_LEVELS.get(r.visibility, 0) <= max_level]
owners_by_id = _resolve_owners_batch(session, visible)
filtered = [
_route_to_read(r)
for r in routes
if VISIBILITY_LEVELS.get(r.visibility, 0) <= max_level
_route_to_read(
r, owner=owners_by_id.get(r.created_by) if r.created_by else None
)
for r in visible
]
return RouteList(items=filtered, total=len(filtered))
@router.post("", response_model=RouteRead, status_code=201)
def create_route(
__: RequireOperatorOrAdmin,
caller: RequireOperatorOrAdmin,
session: DbSession,
body: RouteCreate,
request: Request,
) -> RouteRead:
"""Create a new route (operator or admin)."""
user_id, _ = caller
get_or_create_profile(session, user_id, request)
_assert_visibility_within_role(request, body.visibility)
existing = session.execute(
select(Route).where(
@@ -309,6 +379,7 @@ def create_route(
max_path_length=body.max_path_length,
enabled=body.enabled,
reversible=body.reversible,
created_by=user_id,
)
session.add(route)
session.flush()
@@ -318,7 +389,8 @@ def create_route(
session.refresh(route)
_reevaluate_route(session, route)
invalidate_routes(request)
return _route_to_read(route, quality_avg=None)
owner = _resolve_owner(session, route.created_by)
return _route_to_read(route, quality_avg=None, owner=owner)
@router.get("/{route_id}", response_model=RouteDetail)
@@ -374,7 +446,7 @@ def get_route(
for oid, cnt in contributing.items()
]
read = _route_to_read(route)
read = _route_to_read(route, owner=_resolve_owner(session, route.created_by))
return RouteDetail(
**read.model_dump(),
contributing_observers=contributors,
@@ -566,13 +638,19 @@ def get_route_history(
@router.put("/{route_id}", response_model=RouteRead)
def update_route(
__: RequireOperatorOrAdmin,
caller: RequireOperatorOrAdmin,
session: DbSession,
route_id: str,
body: RouteUpdate,
request: Request,
) -> RouteRead:
"""Update a route (operator or admin)."""
"""Update a route (operator or admin).
Operators may only modify routes they created; admins can modify any
route. Admins claim ownership of legacy (unowned) routes on edit but
do not displace an existing creator.
"""
user_id, _ = caller
route = session.execute(
select(Route).where(Route.id == route_id)
).scalar_one_or_none()
@@ -580,6 +658,11 @@ def update_route(
raise HTTPException(status_code=404, detail="Route not found")
_assert_route_modifiable(request, route)
# Admin claims ownership of legacy (unowned) routes on edit
if resolve_user_role(request) == "admin" and route.created_by is None:
route.created_by = user_id
logger.info("Admin %s claimed ownership of legacy route %s", user_id, route.id)
if body.from_label is not None or body.to_label is not None:
new_from = body.from_label if body.from_label is not None else route.from_label
new_to = body.to_label if body.to_label is not None else route.to_label
@@ -636,7 +719,8 @@ def update_route(
session.refresh(route)
_reevaluate_route(session, route)
invalidate_routes(request)
return _route_to_read(route)
owner = _resolve_owner(session, route.created_by)
return _route_to_read(route, owner=owner)
@router.delete("/{route_id}", status_code=204)
+4
View File
@@ -103,6 +103,10 @@ class Route(Base, UUIDMixin, TimestampMixin):
server_default="true",
nullable=False,
)
created_by: Mapped[Optional[str]] = mapped_column(
String(255),
nullable=True,
)
route_nodes: Mapped[list["RouteNode"]] = relationship(
"RouteNode",
+15
View File
@@ -142,6 +142,17 @@ class RouteUpdate(BaseModel):
return self
class RouteOwner(BaseModel):
"""Creator/owner of a route, for display on route cards."""
user_id: str
name: Optional[str] = None
callsign: Optional[str] = None
profile_id: str
model_config = {"from_attributes": True}
class RouteRead(BaseModel):
"""Schema for reading a route (list-level with lightweight result)."""
@@ -169,6 +180,8 @@ class RouteRead(BaseModel):
"back to ``route_result.quality`` for brand-new routes."
),
)
created_by: Optional[str] = None
owner: Optional[RouteOwner] = None
created_at: datetime
updated_at: datetime
@@ -227,6 +240,8 @@ class RouteDetail(BaseModel):
)
contributing_observers: list[ContributingObserver] = []
recent_matches: list[RecentMatchPath] = []
created_by: Optional[str] = None
owner: Optional[RouteOwner] = None
created_at: datetime
updated_at: datetime
@@ -30,6 +30,8 @@ const ROUTES = {
route_result: { quality: "clear", state: "healthy" },
route_nodes: [],
route_observers: [],
created_by: null,
owner: null,
},
],
};
@@ -135,3 +137,171 @@ describe("Routes role-gated management", () => {
expect(values).not.toContain("admin");
});
});
describe("Routes per-route ownership gating", () => {
function buildConfig(roles: string[], userSub: string) {
return makeConfig({
oidc_enabled: true,
roles,
role_names: { admin: "admin", operator: "operator", member: "member" },
user: { sub: userSub, name: "Test User" },
});
}
afterEach(() => {
window.__APP_CONFIG__ = makeConfig();
});
it("hides edit/delete on routes the operator does not own", async () => {
const cfg = buildConfig(["operator"], "op-1");
window.__APP_CONFIG__ = cfg;
const data = {
items: [
{
...ROUTES.items[0],
id: "other",
from_label: "OtherRoute",
created_by: "different-op",
owner: null,
},
],
};
vi.spyOn(api, "apiGet").mockImplementation(async (path) => {
if (path === "/api/v1/routes") return data;
if (path.match(/\/api\/v1\/routes\/[^/]+$/)) return ROUTE_DETAIL;
if (path.includes("/history")) return ROUTE_HISTORY;
throw new Error(`Unexpected: ${path}`);
});
renderWithProviders(<Routes />, { config: cfg });
await waitFor(() => {
expect(screen.getByText("OtherRoute")).toBeInTheDocument();
});
expect(screen.queryByTestId("edit-route")).toBeNull();
expect(screen.queryByTestId("delete-route")).toBeNull();
});
it("shows edit/delete on routes the operator owns", async () => {
const cfg = buildConfig(["operator"], "op-1");
window.__APP_CONFIG__ = cfg;
const data = {
items: [
{
...ROUTES.items[0],
id: "mine",
from_label: "MyRoute",
created_by: "op-1",
owner: { user_id: "op-1", name: "Test Operator", profile_id: "p1" },
},
],
};
vi.spyOn(api, "apiGet").mockImplementation(async (path) => {
if (path === "/api/v1/routes") return data;
if (path.match(/\/api\/v1\/routes\/[^/]+$/)) return ROUTE_DETAIL;
if (path.includes("/history")) return ROUTE_HISTORY;
throw new Error(`Unexpected: ${path}`);
});
renderWithProviders(<Routes />, { config: cfg });
await waitFor(() => {
expect(screen.queryByTestId("edit-route")).not.toBeNull();
});
expect(screen.getByTestId("edit-route")).toBeInTheDocument();
expect(screen.getByTestId("delete-route")).toBeInTheDocument();
});
it("shows edit/delete on all routes for an admin", async () => {
const cfg = buildConfig(["admin"], "adm-1");
window.__APP_CONFIG__ = cfg;
const data = {
items: [
{
...ROUTES.items[0],
id: "legacy",
from_label: "LegacyRoute",
created_by: null,
owner: null,
},
{
...ROUTES.items[0],
id: "other",
from_label: "OtherRoute",
created_by: "op-99",
owner: { user_id: "op-99", name: "Someone", profile_id: "p2" },
},
],
};
vi.spyOn(api, "apiGet").mockImplementation(async (path) => {
if (path === "/api/v1/routes") return data;
if (path.match(/\/api\/v1\/routes\/[^/]+$/)) return ROUTE_DETAIL;
if (path.includes("/history")) return ROUTE_HISTORY;
throw new Error(`Unexpected: ${path}`);
});
renderWithProviders(<Routes />, { config: cfg });
await waitFor(() => {
expect(screen.getByText("LegacyRoute")).toBeInTheDocument();
});
const editButtons = screen.getAllByTestId("edit-route");
const deleteButtons = screen.getAllByTestId("delete-route");
expect(editButtons).toHaveLength(2);
expect(deleteButtons).toHaveLength(2);
});
it("displays the owner name badge when set", async () => {
const cfg = buildConfig(["operator"], "op-1");
window.__APP_CONFIG__ = cfg;
const data = {
items: [
{
...ROUTES.items[0],
id: "named",
from_label: "NamedRoute",
created_by: "op-1",
owner: { user_id: "op-1", name: "Alice", profile_id: "p1" },
},
],
};
vi.spyOn(api, "apiGet").mockImplementation(async (path) => {
if (path === "/api/v1/routes") return data;
if (path.match(/\/api\/v1\/routes\/[^/]+$/)) return ROUTE_DETAIL;
if (path.includes("/history")) return ROUTE_HISTORY;
throw new Error(`Unexpected: ${path}`);
});
renderWithProviders(<Routes />, { config: cfg });
await waitFor(() => {
expect(screen.getByText("Alice")).toBeInTheDocument();
});
});
it("hides the owner badge when no owner is set", async () => {
const cfg = buildConfig(["operator"], "op-1");
window.__APP_CONFIG__ = cfg;
const data = {
items: [
{
...ROUTES.items[0],
id: "legacy",
from_label: "LegacyRoute",
created_by: null,
owner: null,
},
],
};
vi.spyOn(api, "apiGet").mockImplementation(async (path) => {
if (path === "/api/v1/routes") return data;
if (path.match(/\/api\/v1\/routes\/[^/]+$/)) return ROUTE_DETAIL;
if (path.includes("/history")) return ROUTE_HISTORY;
throw new Error(`Unexpected: ${path}`);
});
renderWithProviders(<Routes />, { config: cfg });
await waitFor(() => {
expect(screen.getByText("LegacyRoute")).toBeInTheDocument();
});
// No owner link should be present (only the route card title contains "LegacyRoute")
const links = screen.queryAllByRole("link");
expect(links.filter((l) => l.textContent === "Test User")).toHaveLength(0);
});
});
@@ -7,7 +7,7 @@ import {
} from "react";
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
import { useTranslation } from "react-i18next";
import { useNavigate } from "react-router";
import { Link, useNavigate } from "react-router";
import { useAppConfig, hasRole } from "@/context/AppConfigContext";
import { apiGet, apiPost, apiPut, apiDelete } from "@/utils/api";
@@ -36,6 +36,7 @@ import {
IconRuler,
IconSatelliteDish,
IconTrash,
IconUser,
} from "@/components/icons";
interface RouteResultInfo {
@@ -58,6 +59,13 @@ interface RouteObserverInfo {
name?: string | null;
}
interface RouteOwnerInfo {
user_id: string;
name?: string | null;
callsign?: string | null;
profile_id: string;
}
interface RouteItem {
id: string;
from_label?: string | null;
@@ -76,6 +84,8 @@ interface RouteItem {
route_result?: RouteResultInfo | null;
route_nodes?: RouteNodeInfo[];
route_observers?: RouteObserverInfo[];
created_by?: string | null;
owner?: RouteOwnerInfo | null;
}
interface RouteListResponse {
@@ -495,14 +505,14 @@ function DetailContent({
function RouteCard({
route,
canManage,
canEdit,
packetsEnabled,
onEdit,
onDelete,
onNavigate,
}: {
route: RouteItem;
canManage: boolean;
canEdit: boolean;
packetsEnabled: boolean;
onEdit: () => void;
onDelete: () => void;
@@ -577,6 +587,21 @@ function RouteCard({
<PathChips route={route} />
</div>
<StatsRow route={route} />
{route.created_by && (
<div className="text-xs opacity-60 flex items-center gap-1 mt-1">
<IconUser className="h-3 w-3" />
{route.owner ? (
<Link
to={`/profile/${route.owner.profile_id}`}
className="link link-hover"
>
{route.owner.name || route.owner.user_id}
</Link>
) : (
<span>{route.created_by}</span>
)}
</div>
)}
{detail ? (
<DetailContent
route={route}
@@ -590,7 +615,7 @@ function RouteCard({
<span className="loading loading-spinner loading-sm opacity-50"></span>
</div>
)}
{canManage && (
{canEdit && (
<div className="flex gap-2 mt-auto pt-2">
<button
className="btn btn-xs btn-outline"
@@ -1159,6 +1184,10 @@ export function RoutesPage() {
const config = useAppConfig();
const packetsEnabled = config.features?.packets !== false;
const canManage = hasRole("admin") || hasRole("operator");
const currentUserId = config.user?.sub;
const isAdmin = hasRole("admin");
const canEditRoute = (r: RouteItem) =>
isAdmin || (!!r.created_by && r.created_by === currentUserId);
usePageTitle("routes.title");
const queryClient = useQueryClient();
@@ -1516,7 +1545,7 @@ export function RoutesPage() {
<RouteCard
key={r.id}
route={r}
canManage={canManage}
canEdit={canEditRoute(r)}
packetsEnabled={packetsEnabled}
onEdit={() => openEditModal(r)}
onDelete={() => openDeleteModal(r)}
+350 -7
View File
@@ -11,6 +11,7 @@ from meshcore_hub.common.models import (
Route,
RouteNode,
)
from meshcore_hub.common.models.user_profile import UserProfile
ADMIN_HEADERS = {"X-User-Id": "test-admin", "X-User-Roles": "admin"}
OPERATOR_HEADERS = {"X-User-Id": "test-operator", "X-User-Roles": "operator"}
@@ -1319,7 +1320,16 @@ class TestPrecomputedRecentMatches:
class TestRouteOperatorPermissions:
"""Operators may manage routes but cannot scope above their own role."""
"""Operators may manage routes but cannot scope above their own role.
Ownership model (PR #337):
- Operators can edit/delete only routes they created.
- Admins can edit/delete any route; they claim ownership of legacy
(unowned) routes on edit but do not displace an existing creator.
- Routes with NULL ``created_by`` (legacy) are admin-only.
"""
# ── Create: visibility + created_by stamping ──────────────────────
def test_operator_create_at_own_level(self, client_no_auth, api_db_session):
nodes = _sample_nodes(api_db_session)
@@ -1388,8 +1398,50 @@ class TestRouteOperatorPermissions:
assert resp.status_code == 201
assert resp.json()["visibility"] == "admin"
def test_operator_update_own_level(self, client_no_auth, api_db_session):
route = Route(from_label="Op", to_label="End", visibility="operator")
def test_create_stamps_created_by(self, client_no_auth, api_db_session):
"""Route creation stamps the caller's user_id into created_by."""
nodes = _sample_nodes(api_db_session)
api_db_session.commit()
resp = client_no_auth.post(
"/api/v1/routes",
json={
"from_label": "Owner",
"to_label": "End",
"visibility": "operator",
"node_public_keys": [n.public_key for n in nodes],
},
headers=OPERATOR_HEADERS,
)
assert resp.status_code == 201
assert resp.json()["created_by"] == "test-operator"
def test_admin_create_stamps_created_by(self, client_no_auth, api_db_session):
nodes = _sample_nodes(api_db_session)
api_db_session.commit()
resp = client_no_auth.post(
"/api/v1/routes",
json={
"from_label": "AdmOwner",
"to_label": "End",
"visibility": "admin",
"node_public_keys": [n.public_key for n in nodes],
},
headers=ADMIN_HEADERS,
)
assert resp.status_code == 201
assert resp.json()["created_by"] == "test-admin"
# ── Update: ownership enforcement ─────────────────────────────────
def test_operator_update_own_route(self, client_no_auth, api_db_session):
route = Route(
from_label="Op",
to_label="End",
visibility="operator",
created_by="test-operator",
)
api_db_session.add(route)
api_db_session.commit()
@@ -1402,7 +1454,12 @@ class TestRouteOperatorPermissions:
assert resp.json()["description"] == "edited by operator"
def test_operator_update_community_route(self, client_no_auth, api_db_session):
route = Route(from_label="Pub", to_label="End", visibility="community")
route = Route(
from_label="Pub",
to_label="End",
visibility="community",
created_by="test-operator",
)
api_db_session.add(route)
api_db_session.commit()
@@ -1414,6 +1471,7 @@ class TestRouteOperatorPermissions:
assert resp.status_code == 200
def test_operator_update_admin_route_404(self, client_no_auth, api_db_session):
"""Above-tier routes yield 404 (existence hidden)."""
route = Route(from_label="Secret", to_label="End", visibility="admin")
api_db_session.add(route)
api_db_session.commit()
@@ -1425,10 +1483,94 @@ class TestRouteOperatorPermissions:
)
assert resp.status_code == 404
def test_operator_update_other_operator_route_403(
self, client_no_auth, api_db_session
):
"""Operator cannot edit a route created by another operator."""
route = Route(
from_label="Other",
to_label="End",
visibility="operator",
created_by="different-operator",
)
api_db_session.add(route)
api_db_session.commit()
resp = client_no_auth.put(
f"/api/v1/routes/{route.id}",
json={"description": "attempt"},
headers=OPERATOR_HEADERS,
)
assert resp.status_code == 403
def test_operator_update_legacy_null_route_403(
self, client_no_auth, api_db_session
):
"""Routes with NULL created_by (legacy) are admin-only."""
route = Route(
from_label="Legacy",
to_label="End",
visibility="operator",
created_by=None,
)
api_db_session.add(route)
api_db_session.commit()
resp = client_no_auth.put(
f"/api/v1/routes/{route.id}",
json={"description": "attempt"},
headers=OPERATOR_HEADERS,
)
assert resp.status_code == 403
def test_operator_update_admin_created_route_403(
self, client_no_auth, api_db_session
):
"""Operator cannot edit a route created by an admin (even if visible)."""
route = Route(
from_label="AdminMade",
to_label="End",
visibility="operator",
created_by="test-admin",
)
api_db_session.add(route)
api_db_session.commit()
resp = client_no_auth.put(
f"/api/v1/routes/{route.id}",
json={"description": "attempt"},
headers=OPERATOR_HEADERS,
)
assert resp.status_code == 403
def test_admin_update_any_route(self, client_no_auth, api_db_session):
"""Admin can update a route regardless of who created it."""
route = Route(
from_label="Any",
to_label="End",
visibility="operator",
created_by="someone-else",
)
api_db_session.add(route)
api_db_session.commit()
resp = client_no_auth.put(
f"/api/v1/routes/{route.id}",
json={"description": "admin edit"},
headers=ADMIN_HEADERS,
)
assert resp.status_code == 200
assert resp.json()["description"] == "admin edit"
def test_operator_escalate_visibility_rejected(
self, client_no_auth, api_db_session
):
route = Route(from_label="Op", to_label="End", visibility="operator")
route = Route(
from_label="Op",
to_label="End",
visibility="operator",
created_by="test-operator",
)
api_db_session.add(route)
api_db_session.commit()
@@ -1439,8 +1581,87 @@ class TestRouteOperatorPermissions:
)
assert resp.status_code == 403
def test_operator_delete_own_level(self, client_no_auth, api_db_session):
route = Route(from_label="Op", to_label="End", visibility="operator")
def test_operator_cannot_claim_ownership_via_body(
self, client_no_auth, api_db_session
):
"""created_by in the PUT body is ignored (Pydantic extra='ignore')."""
route = Route(
from_label="Hijack",
to_label="End",
visibility="operator",
created_by="test-operator",
)
api_db_session.add(route)
api_db_session.commit()
resp = client_no_auth.put(
f"/api/v1/routes/{route.id}",
json={"description": "ok", "created_by": "attacker"},
headers=OPERATOR_HEADERS,
)
assert resp.status_code == 200
assert resp.json()["created_by"] == "test-operator"
# ── Admin ownership semantics ─────────────────────────────────────
def test_admin_edit_preserves_existing_ownership(
self, client_no_auth, api_db_session
):
"""Admin editing an operator-created route does NOT take ownership."""
route = Route(
from_label="Preserve",
to_label="End",
visibility="operator",
created_by="test-operator",
)
api_db_session.add(route)
api_db_session.commit()
resp = client_no_auth.put(
f"/api/v1/routes/{route.id}",
json={"description": "admin tweaked it"},
headers=ADMIN_HEADERS,
)
assert resp.status_code == 200
# Ownership stays with the original creator
assert resp.json()["created_by"] == "test-operator"
# Original operator can still edit
resp2 = client_no_auth.put(
f"/api/v1/routes/{route.id}",
json={"description": "operator still owns it"},
headers=OPERATOR_HEADERS,
)
assert resp2.status_code == 200
def test_admin_edit_adopts_null_created_by(self, client_no_auth, api_db_session):
"""Admin editing a legacy route (NULL created_by) takes ownership."""
route = Route(
from_label="Adopt",
to_label="End",
visibility="operator",
created_by=None,
)
api_db_session.add(route)
api_db_session.commit()
resp = client_no_auth.put(
f"/api/v1/routes/{route.id}",
json={"description": "adopted"},
headers=ADMIN_HEADERS,
)
assert resp.status_code == 200
assert resp.json()["created_by"] == "test-admin"
# ── Delete: ownership enforcement ─────────────────────────────────
def test_operator_delete_own_route(self, client_no_auth, api_db_session):
route = Route(
from_label="Del",
to_label="End",
visibility="operator",
created_by="test-operator",
)
api_db_session.add(route)
api_db_session.commit()
@@ -1460,3 +1681,125 @@ class TestRouteOperatorPermissions:
headers=OPERATOR_HEADERS,
)
assert resp.status_code == 404
def test_operator_delete_other_route_403(self, client_no_auth, api_db_session):
"""Operator cannot delete a route created by someone else."""
route = Route(
from_label="Other",
to_label="End",
visibility="operator",
created_by="different-operator",
)
api_db_session.add(route)
api_db_session.commit()
resp = client_no_auth.delete(
f"/api/v1/routes/{route.id}",
headers=OPERATOR_HEADERS,
)
assert resp.status_code == 403
def test_operator_delete_legacy_null_route_403(
self, client_no_auth, api_db_session
):
"""Operator cannot delete a legacy route with NULL created_by."""
route = Route(
from_label="Legacy",
to_label="End",
visibility="operator",
created_by=None,
)
api_db_session.add(route)
api_db_session.commit()
resp = client_no_auth.delete(
f"/api/v1/routes/{route.id}",
headers=OPERATOR_HEADERS,
)
assert resp.status_code == 403
def test_admin_delete_any_route(self, client_no_auth, api_db_session):
"""Admin can delete a route regardless of who created it."""
route = Route(
from_label="AnyDel",
to_label="End",
visibility="operator",
created_by="someone-else",
)
api_db_session.add(route)
api_db_session.commit()
resp = client_no_auth.delete(
f"/api/v1/routes/{route.id}",
headers=ADMIN_HEADERS,
)
assert resp.status_code == 204
# ── Owner name resolution in responses ────────────────────────────
def test_owner_name_resolved_in_list(self, client_no_auth, api_db_session):
"""List includes owner friendly name when a profile exists, null otherwise."""
profile = UserProfile(
user_id="test-operator",
name="Alice Operator",
)
api_db_session.add(profile)
named = Route(
from_label="Named",
to_label="End",
visibility="operator",
created_by="test-operator",
)
ghost = Route(
from_label="Ghost",
to_label="End",
visibility="operator",
created_by="ghost-user",
)
api_db_session.add_all([named, ghost])
api_db_session.commit()
resp = client_no_auth.get(
"/api/v1/routes",
headers=OPERATOR_HEADERS,
)
assert resp.status_code == 200
items = resp.json()["items"]
named_match = [r for r in items if r["from_label"] == "Named"]
assert len(named_match) == 1
assert named_match[0]["owner"] is not None
assert named_match[0]["owner"]["name"] == "Alice Operator"
assert named_match[0]["owner"]["user_id"] == "test-operator"
ghost_match = [r for r in items if r["from_label"] == "Ghost"]
assert len(ghost_match) == 1
assert ghost_match[0]["created_by"] == "ghost-user"
assert ghost_match[0]["owner"] is None
def test_owner_resolved_in_detail(self, client_no_auth, api_db_session):
"""GET /routes/{id} includes owner info."""
profile = UserProfile(
user_id="test-operator",
name="Bob",
callsign="VK1BOB",
)
api_db_session.add(profile)
route = Route(
from_label="Detail",
to_label="End",
visibility="operator",
created_by="test-operator",
)
api_db_session.add(route)
api_db_session.commit()
resp = client_no_auth.get(
f"/api/v1/routes/{route.id}",
headers=OPERATOR_HEADERS,
)
assert resp.status_code == 200
body = resp.json()
assert body["owner"] is not None
assert body["owner"]["name"] == "Bob"
assert body["owner"]["callsign"] == "VK1BOB"