Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion VERSION
Original file line number Diff line number Diff line change
@@ -1 +1 @@
0.1.564
0.1.568
52 changes: 52 additions & 0 deletions fleet_platform/api/limiter.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,9 @@
from typing import Callable

from slowapi import Limiter
from starlette.datastructures import MutableHeaders
from starlette.requests import Request
from starlette.types import ASGIApp, Message, Receive, Scope, Send


def make_real_ip_key(proxy_count: int) -> Callable[[Request], str]:
Expand Down Expand Up @@ -69,3 +71,53 @@ def _default_key(request: Request) -> str:


limiter = Limiter(key_func=_default_key)


class RateLimitHeadersMiddleware:
"""Emit ``X-RateLimit-*`` headers on every HTTP response for rate-limited routes.

SlowAPI's own header injection is not usable here:

* ``Limiter(headers_enabled=True)`` makes the ``@limiter.limit`` *decorator*
try to inject headers into the endpoint's return value. Our endpoints
return plain dicts (not ``Response`` objects) and don't declare a
``response: Response`` parameter, so that path raises
"parameter `response` must be an instance of Response" — it would break
every decorated dict-returning route.
* ``SlowAPIMiddleware`` deliberately *exempts* any route that carries a
``@limiter.limit`` decorator, so it never adds headers to e.g. ``/auth/login``.

This is a pure-ASGI middleware (not ``BaseHTTPMiddleware``) so it leaves
WebSocket scopes completely untouched — wrapping WebSocket routes in
``BaseHTTPMiddleware`` breaks their close-code handshake. For HTTP requests
we read ``view_rate_limit`` from the shared scope state (the decorator sets
it in ``_check_request_limit`` *before* the endpoint runs, so it is present
even when the endpoint raises, e.g. a 401 login) and append the standard
headers as the response starts, mirroring ``Limiter._inject_headers``.
"""

def __init__(self, app: ASGIApp) -> None:
self.app = app

async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] != "http":
await self.app(scope, receive, send)
return

async def send_wrapper(message: Message) -> None:
if message["type"] == "http.response.start":
current_limit = (scope.get("state") or {}).get("view_rate_limit")
limiter_obj = getattr(getattr(scope.get("app"), "state", None), "limiter", None)
if current_limit is not None and limiter_obj is not None:
try:
item, args = current_limit
window_stats = limiter_obj.limiter.get_window_stats(item, *args)
headers = MutableHeaders(scope=message)
headers["X-RateLimit-Limit"] = str(item.amount)
headers["X-RateLimit-Remaining"] = str(window_stats[1])
headers["X-RateLimit-Reset"] = str(1 + window_stats[0])
except Exception: # never let header decoration break a response
pass
await send(message)

await self.app(scope, receive, send_wrapper)
7 changes: 6 additions & 1 deletion fleet_platform/api/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
from slowapi.errors import RateLimitExceeded
from starlette.responses import Response

from fleet_platform.api.limiter import limiter
from fleet_platform.api.limiter import RateLimitHeadersMiddleware, limiter
from fleet_platform.api.metrics_auth import verify_metrics_request
from fleet_platform.api.metrics_collectors import refresh_all_gauges
from fleet_platform.api.routes import (
Expand Down Expand Up @@ -164,6 +164,11 @@ async def structured_validation_exception_handler(request: Request, exc: Request

app.add_middleware(SecurityHeaderMiddleware)

# Adds X-RateLimit-* headers to every response on rate-limited routes (see
# RateLimitHeadersMiddleware docstring for why SlowAPI's own injection can't
# be used with our dict-returning decorated endpoints).
app.add_middleware(RateLimitHeadersMiddleware)

app.include_router(health.router, tags=["health"])
app.include_router(auth.router, tags=["auth"])
app.include_router(nodes.router, tags=["nodes"])
Expand Down
7 changes: 6 additions & 1 deletion fleet_platform/api/routes/fleet.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
from fleet_platform.api.deps import get_db, get_redis
from fleet_platform.core.audit import audit
from fleet_platform.core.auth import get_current_user, hash_password, require_role
from fleet_platform.core.validators import MINION_ID_RE
from fleet_platform.models.node import Node
from fleet_platform.schemas.fleet import FleetOverviewResponse
from fleet_platform.schemas.node_import import (
Expand Down Expand Up @@ -194,7 +195,11 @@ async def import_commit(
_ssh_key = payload.ssh_key or None
_auth_mode = payload.ssh_auth_mode or ("key" if (_ssh_key and not _ssh_pw) else "password")

new_rows = [r for r in payload.rows if r.status == "new"]
# Commit only acts on rows the validate step marked "new". Re-check the
# minion_id format defensively here too: a client could POST status="new"
# with a malformed minion_id directly, bypassing the validate endpoint.
# Such rows are skipped (never persisted) rather than 422-ing the whole batch.
new_rows = [r for r in payload.rows if r.status == "new" and MINION_ID_RE.match(r.minion_id or "")]
skipped = len(payload.rows) - len(new_rows)
created_ids: list[str] = []
created_nodes: list[Node] = []
Expand Down
14 changes: 9 additions & 5 deletions fleet_platform/schemas/node_import.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,18 @@
# fleet_platform/schemas/node_import.py
"""Pydantic schemas for bulk node import endpoints (#360)."""

from pydantic import BaseModel, field_validator

from fleet_platform.core.validators import validate_minion_id
from pydantic import BaseModel


class ImportRow(BaseModel):
# NOTE: minion_id is intentionally NOT hard-validated here. ImportRow is the
# transport for the dry-run *validate* response, which must be able to
# represent rows that FAILED validation (status="invalid") so the UI can
# show the user exactly which rows were rejected and why. Raising at parse
# time would 500 the validate endpoint instead of reporting invalid rows.
# Format enforcement happens in node_import.validate_row (soft, categorising)
# and again defensively on the commit path (see fleet.import_commit), so a
# bad minion_id can never be persisted.
minion_id: str
hostname: str | None = None
ip: str | None = None
Expand All @@ -15,8 +21,6 @@ class ImportRow(BaseModel):
status: str = "new"
reason: str = ""

_validate_minion_id = field_validator("minion_id")(validate_minion_id)


class ImportValidateRequest(BaseModel):
source: str
Expand Down
9 changes: 9 additions & 0 deletions fleet_platform/services/node_import.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
import ipaddress
import re

from fleet_platform.core.validators import MINION_ID_RE

_HOSTNAME_RE = re.compile(r"^[a-zA-Z0-9]([a-zA-Z0-9\-_.]{0,253}[a-zA-Z0-9])?$")


Expand Down Expand Up @@ -77,6 +79,13 @@ def validate_row(row: dict, existing_minions: set, existing_ips: set) -> dict:
if not mid:
return {**out, "status": "invalid", "reason": "missing minion_id"}

if not MINION_ID_RE.match(mid):
return {
**out,
"status": "invalid",
"reason": "minion_id must be 1-128 chars, only [a-zA-Z0-9._-] allowed",
}

if ip:
try:
ipaddress.ip_address(ip)
Expand Down
2 changes: 1 addition & 1 deletion frontend/package.json
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
{
"name": "frontend",
"private": true,
"version": "0.1.564",
"version": "0.1.568",
"packageManager": "[email protected]",
"type": "module",
"scripts": {
Expand Down
7 changes: 6 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,12 @@ dev = [
"bandit[toml]>=1.9.4",
"pre-commit>=4.6.0",
"hypothesis>=6.155.2",
"mutmut>=3.1.0",
# Pin below 3.6.0: that release renamed the [tool.mutmut] keys
# (paths_to_mutate->source_paths, tests_dir folded into
# pytest_add_cli_args_test_selection) AND shipped a config-loader bug that
# crashes `mutmut results` with "can only concatenate list (not 'str') to
# list", failing the mutation CI job. 3.5.0 reads our string-valued config.
"mutmut>=3.1.0,<3.6.0",
]

# Only package fleet_platform — exclude playbooks/ and other top-level dirs.
Expand Down
34 changes: 24 additions & 10 deletions tests/integration/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,20 +48,26 @@ async def db_session(test_engine):
async def app_with_test_db(test_engine):
from unittest.mock import AsyncMock, MagicMock

from slowapi import Limiter
from slowapi.util import get_remote_address

import fleet_platform.api.limiter as limiter_module

# Use in-memory rate limiter for tests to avoid 429 false positives
test_limiter = Limiter(key_func=get_remote_address, storage_uri="memory://")
limiter_module.limiter = test_limiter

from fleet_platform.api import deps
from fleet_platform.api.main import create_app

# Do NOT swap limiter_module.limiter for a throwaway instance here. The route
# @limiter.limit decorators bind to whichever limiter object the routes
# imported at first load; reassigning the module attribute afterwards cannot
# rebind them and only desynchronises app.state.limiter from the limiter the
# decorators actually use — which is exactly what broke the behavioural
# rate-limit tests (they reset limiter_module.limiter, but the decorators
# were firing on a stale per-module test_limiter). Instead, share the real
# production limiter and just zero its in-memory storage so hit counters from
# other modules don't bleed in and cause spurious 429s.
try:
limiter_module.limiter._storage.reset()
except Exception:
pass

app = create_app()
app.state.limiter = test_limiter
app.state.limiter = limiter_module.limiter
TestSession = async_sessionmaker(test_engine, expire_on_commit=False)

async def override_get_db():
Expand All @@ -86,7 +92,15 @@ async def override_get_redis():
app.dependency_overrides[deps.get_db] = override_get_db
app.dependency_overrides[deps.get_redis] = override_get_redis
app._test_mock_redis = mock_redis # expose for tests that need to configure it
return app
try:
yield app
finally:
# Clear any hits the route decorators accumulated on the shared limiter
# during this module so later modules start from a clean counter.
try:
limiter_module.limiter._storage.reset()
except Exception:
pass


@pytest_asyncio.fixture(scope="module", loop_scope="module")
Expand Down
18 changes: 8 additions & 10 deletions tests/integration/test_agent_run_795.py
Original file line number Diff line number Diff line change
Expand Up @@ -267,23 +267,21 @@ async def test_list_agent_actions_returns_list(admin_client: AsyncClient):

async def test_reject_agent_action_over_http(admin_client: AsyncClient, db_session: AsyncSession):
"""POST /agent/actions/{id}/reject transitions the action to 'rejected'."""
from sqlalchemy import select

from fleet_platform.models.pending_action import PendingAction

action = await _create_pending_action(db_session, requested_by="[email protected]")
action_id = str(action.id)

resp = await admin_client.post(f"/api/v1/agent/actions/{action_id}/reject")
assert resp.status_code == 200
assert resp.json().get("status") == "rejected"

# DB must reflect the rejected status
refreshed = (
await db_session.execute(select(PendingAction).where(PendingAction.id == action.id))
).scalar_one_or_none()
assert refreshed is not None
assert refreshed.status == "rejected"
# DB must reflect the rejected status. The API ran the update in its own
# session; this test's db_session (expire_on_commit=False) still holds the
# original instance in its identity map, so refresh it to read the committed
# row rather than asserting against the stale cached object. refresh() is
# awaited, so it is safe in the async session (unlike touching an expired
# attribute, which would trigger a sync lazy-load).
await db_session.refresh(action)
assert action.status == "rejected"


async def test_approve_agent_action_over_http(admin_client: AsyncClient, db_session: AsyncSession, admin_user):
Expand Down
19 changes: 11 additions & 8 deletions tests/integration/test_auth_endpoints.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,19 +69,22 @@ def auth_fake_redis() -> _FakeRedis:

@pytest_asyncio.fixture(loop_scope="module")
async def auth_client(test_engine, auth_fake_redis: _FakeRedis):
from slowapi import Limiter
from slowapi.util import get_remote_address

import fleet_platform.api.limiter as limiter_module

test_limiter = Limiter(key_func=get_remote_address, storage_uri="memory://")
limiter_module.limiter = test_limiter

from fleet_platform.api import deps
from fleet_platform.api.main import create_app

# Share the real production limiter (do not swap in a throwaway instance —
# the route decorators bind to the limiter present at first import and can't
# be rebound, which desynchronises app.state.limiter from the limiter the
# decorators use and breaks the behavioural rate-limit tests). Just zero its
# in-memory storage so counts from other modules don't cause spurious 429s.
try:
limiter_module.limiter._storage.reset()
except Exception:
pass

app = create_app()
app.state.limiter = test_limiter
app.state.limiter = limiter_module.limiter
TestSession = async_sessionmaker(test_engine, expire_on_commit=False)

async def override_get_db():
Expand Down
19 changes: 11 additions & 8 deletions tests/integration/test_check_minion_id.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,19 +37,22 @@ async def db_session(test_engine):

@pytest_asyncio.fixture(scope="module", loop_scope="module")
async def app_with_test_db(test_engine):
from slowapi import Limiter
from slowapi.util import get_remote_address

import fleet_platform.api.limiter as limiter_module

test_limiter = Limiter(key_func=get_remote_address, storage_uri="memory://")
limiter_module.limiter = test_limiter

from fleet_platform.api import deps
from fleet_platform.api.main import create_app

# Share the real production limiter (do not swap in a throwaway instance —
# the route decorators bind to the limiter present at first import and can't
# be rebound, which desynchronises app.state.limiter from the limiter the
# decorators use and breaks the behavioural rate-limit tests). Just zero its
# in-memory storage so counts from other modules don't cause spurious 429s.
try:
limiter_module.limiter._storage.reset()
except Exception:
pass

app = create_app()
app.state.limiter = test_limiter
app.state.limiter = limiter_module.limiter
TestSession = async_sessionmaker(test_engine, expire_on_commit=False)

async def override_get_db():
Expand Down
19 changes: 11 additions & 8 deletions tests/integration/test_node_import_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,19 +47,22 @@ async def db_session(test_engine):
async def app_with_test_db(test_engine):
from unittest.mock import AsyncMock

from slowapi import Limiter
from slowapi.util import get_remote_address

import fleet_platform.api.limiter as limiter_module

test_limiter = Limiter(key_func=get_remote_address, storage_uri="memory://")
limiter_module.limiter = test_limiter

from fleet_platform.api import deps
from fleet_platform.api.main import create_app

# Share the real production limiter (do not swap in a throwaway instance —
# the route decorators bind to the limiter present at first import and can't
# be rebound, which desynchronises app.state.limiter from the limiter the
# decorators use and breaks the behavioural rate-limit tests). Just zero its
# in-memory storage so counts from other modules don't cause spurious 429s.
try:
limiter_module.limiter._storage.reset()
except Exception:
pass

app = create_app()
app.state.limiter = test_limiter
app.state.limiter = limiter_module.limiter
TestSession = async_sessionmaker(test_engine, expire_on_commit=False)

async def override_get_db():
Expand Down
11 changes: 8 additions & 3 deletions tests/unit/test_ist_timestamps.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,11 @@ def test_ist_utils_file_exists():
def test_formatist_is_exported():
content = open("frontend/src/utils/time.ts").read()
assert "export function formatIST" in content
assert "Asia/Kolkata" in content
assert "IST" in content
# #796/#877: timestamps now render in the viewer's local timezone via
# getTimezone() (with timeZoneName for the suffix) rather than a hardcoded
# Asia/Kolkata zone.
assert "getTimezone()" in content
assert "timeZoneName" in content


def test_node_detail_uses_formatist():
Expand All @@ -26,4 +29,6 @@ def test_node_detail_uses_formatist():

def test_ist_suffix_in_utility():
content = open("frontend/src/utils/time.ts").read()
assert "+ ' IST'" in content or "IST'" in content
# The timezone suffix is now derived dynamically from the viewer's locale
# (timeZoneName: 'short') instead of a hardcoded ' IST' string (#796/#877).
assert "timeZoneName" in content
Loading
Loading