Files
MartinNYHC 5da403ba0c Feature/makerworld (#1099)
* feat(makerworld): URL-paste import and print for MakerWorld models

  Add a dedicated /makerworld sidebar page where users paste a MakerWorld
  model URL and get the full plate list + one-click "Import to Library" or
  "Print Now". Closes the workflow gap that kept LAN-only users on the
  Bambu Handy app solely for MakerWorld download-and-send.

  The authenticated tier reuses the existing Bambu Cloud token that
  Bambuddy already stores for firmware checks and slicer settings --
  MakerWorld shares the same auth backend, so the same JWT works there.
  No separate OAuth flow, no companion browser extension, no credential
  hijack. Anonymous users can still paste a URL and see model metadata;
  the 3MF download itself requires the Cloud login.

  Print Now hands off to the existing PrintModal (plate picker + AMS
  mapping + dispatch) so multi-filament models work via the same code
  path as library-file prints. Imported 3MFs are stored through a new
  shared save_3mf_bytes_to_library() helper so the multipart upload
  route and the MakerWorld import route don't duplicate 3MF parsing +
  thumbnail extraction logic.

  LibraryFile gains indexed source_type + source_url columns. Re-pasting
  a URL for a model already in the library returns the existing row
  instead of re-downloading -- dedupe is by canonicalised URL, not SHA256,
  because MakerWorld's download URLs are signed and change per request.

  Thumbnail proxy (/makerworld/thumbnail) hot-links through the backend
  instead of directly to makerworld.bblmw.com -- the SPA's img-src CSP
  stays strict and users' IPs don't hit MakerWorld's CDN logs. The
  endpoint is intentionally unauthenticated since <img> tags can't carry
  a Bearer token; SSRF-guarded by a CDN host allowlist so it can't be
  used as a generic proxy.

  Search and browse-catalogue are explicitly out of scope. The public
  design/search endpoint returns empty results from server-originated
  requests (likely needs csrf/session state reproducible only from a
  real browser), and the __NEXT_DATA__ HTML fallback is blocked by
  Cloudflare. URL-paste covers the realistic discovery pattern (Reddit /
  YouTube / shared links).

  Headers match kloshi-io/makerworld-api-reverse's production-tested set
  (User-Agent: 3d-printing-service/1.0, x-bbl-* client identifiers,
  Referer). The /instance/{id}/f3mf call includes ?type=download which
  community userscripts use to signal legitimate download intent. 418
  responses (MakerWorld's CAPTCHA gate) retry once with backoff and then
  surface a clear actionable error with an "Open on MakerWorld" fallback
  link; we never try to evade bot detection.

  Permissions: new makerworld:view (browse metadata, view thumbnails) and
  makerworld:import (save 3MFs to library). Administrators and Operators
  get both; Viewers get view-only. Migration grants these to existing
  groups based on whether they already have library:upload / library:read.

  Disclaimer in the UI and wiki page mirrors kloshi's framing: not
  affiliated with or endorsed by MakerWorld or Bambu Lab, interoperability
  only, not intended to circumvent access controls.

  Tests: 30 backend (service + routes) + 4 frontend. Full backend suite
  (1931 tests) clean. Frontend build clean.

* feat(makerworld): ship working URL-paste import via api.bambulab.com iot-service

  The MakerWorld integration shipped in 0.2.4b1 dev was broken for most
  public models: the makerworld.com/design-service path returns "Please
  log in to download models" even with a valid Bambu Cloud bearer,
  because it's cookie-gated behind Cloudflare. Published reverse-
  engineering projects work around this by pasting browser cookies; we
  route around it entirely by using the api.bambulab.com/iot-service
  endpoint (documented by Pr0zak/YASTL#51), which accepts the same
  bearer Bambuddy already has and returns a presigned S3 URL.

  Working flow:
    GET api.bambulab.com/v1/design-service/design/{id}  → metadata
    GET api.bambulab.com/v1/iot-service/api/user/profile/{pid}?model_id=<str>
         Authorization: Bearer {cloud_token}             → signed S3 URL
    urllib.request (no redirects, no query re-encoding)  → bytes

  Notes on each step:
    - The model_id query param is the alphanumeric string from the
      design response (e.g. US2bb73b106683e5), NOT the integer designId
      from the /models/{N} URL. The import route fetches design metadata
      first to get it.
    - S3 presigned URLs MUST be fetched with urllib (not httpx/curl_cffi)
      because the signature is computed over exact query-string bytes;
      any normalising encoder breaks it with SignatureDoesNotMatch 400s
      (YASTL#52 hit the same issue). Wrapped in a no-redirect opener so
      the .amazonaws.com host allowlist guarantee isn't bypassed by a
      302 elsewhere.
    - The canonical source_url now includes profile_id so different
      plates of the same model get distinct library entries. Older rows
      from dev builds keep the model-level URL; the resolve endpoint's
      "already imported" check LIKEs both shapes.

  UI rebuild:
    - Per-plate Save + Save & Slice in Bambu Studio / OrcaSlicer (the
      plate is unsliced source, so "Print Now" was misleading and is
      replaced by an explicit slicer hand-off).
    - Import all plates with sequential progress.
    - Folder picker (default: auto-created top-level "MakerWorld"
      folder, created on first import, folder tree invalidated so
      File Manager shows it immediately).
    - Image gallery per plate with keyboard-navigable lightbox.
    - Recent imports sidebar (sticky on lg+, vertical list with
      jump-to-library / slicer / open-on-makerworld icons).
    - Inline follow-up actions on imported plate rows so the user
      doesn't scroll back to a top-of-page card.
    - Per-plate delete via the standard ConfirmModal (no window.confirm).
    - Elapsed-time + phase label during import so the 10-30s synchronous
      POST doesn't feel frozen.
    - URL-change detection drops the preview when the pasted URL
      diverges from the resolved one.

  Security hardening (found in review):
    - DOMPurify.sanitize on the MakerWorld HTML summary before
      dangerouslySetInnerHTML (user-authored content).
    - <img> tags in that HTML routed through the thumbnail proxy so
      the SPA's img-src 'self' data: blob: CSP isn't widened.
    - /makerworld/thumbnail uses follow_redirects=False (the host
      allowlist only covers the initial URL).
    - 3MF CDN fetch strips the bearer (signed URL is the credential).
    - S3 fetch uses a no-op HTTPRedirectHandler for the same reason.
    - Upstream filename is os.path.basename'd before persisting.

  Tests: 46 backend service unit tests, 19 route tests, 12 frontend
  tests — all passing. All user-facing strings localised across the
  8 UI languages.

* - frontend/src/App.tsx — removed the 3 stale <AdminRoute> lines (kept the 3 <PermissionRoute> equivalents). TSC + Vite both clean.
  - backend/tests/integration/test_auth_api.py — added # pragma: allowlist secret + # noqa: S106 on the test fixture line that GitGuardian flagged.
2026-04-23 14:10:14 +02:00

970 lines
35 KiB
Python

"""Integration tests for Authentication API endpoints.
Tests the full request/response cycle for /api/v1/auth/ and /api/v1/users/ endpoints.
"""
import pytest
from httpx import AsyncClient
class TestAuthStatusAPI:
"""Integration tests for /api/v1/auth/status endpoint."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_auth_status_disabled(self, async_client: AsyncClient):
"""Verify auth status returns disabled when not configured."""
response = await async_client.get("/api/v1/auth/status")
assert response.status_code == 200
result = response.json()
assert "auth_enabled" in result
assert result["auth_enabled"] is False
assert result["requires_setup"] is True
class TestAuthSetupAPI:
"""Integration tests for /api/v1/auth/setup endpoint."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_setup_auth_disabled(self, async_client: AsyncClient):
"""Verify auth can be set up with auth disabled (no password required)."""
response = await async_client.post(
"/api/v1/auth/setup",
json={"auth_enabled": False},
)
assert response.status_code == 200
result = response.json()
assert result["auth_enabled"] is False
assert result["admin_created"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_setup_auth_enabled_requires_credentials(self, async_client: AsyncClient):
"""Verify enabling auth requires admin username and password."""
response = await async_client.post(
"/api/v1/auth/setup",
json={"auth_enabled": True},
)
assert response.status_code == 400
assert "Admin username and password are required" in response.json()["detail"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_setup_auth_enabled_with_credentials(self, async_client: AsyncClient):
"""Verify auth can be enabled with admin credentials."""
response = await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "testadmin",
"admin_password": "TestPass1!",
},
)
assert response.status_code == 200
result = response.json()
assert result["auth_enabled"] is True
assert result["admin_created"] is True
@pytest.mark.asyncio
@pytest.mark.integration
async def test_setup_weak_password_rejected_when_creating_new_admin(self, async_client: AsyncClient):
"""Complexity is enforced only when a new admin is being created."""
response = await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "weakpw_admin",
"admin_password": "NoSpecial1",
},
)
assert response.status_code == 400
assert "special character" in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_setup_reenable_with_existing_admin_ignores_password(self, async_client: AsyncClient, db_session):
"""Re-enabling auth when an admin already exists must not reject the placeholder
password the frontend still sends. Regression for the LDAP re-enable flow that
previously 422'd because the Pydantic schema enforced complexity unconditionally.
"""
from backend.app.core.auth import get_password_hash
from backend.app.models.user import User
existing = User(
username="existing_admin",
# pragma: allowlist secret — test fixture only, not a real credential
password_hash=get_password_hash("DoesNotMatter1!"), # noqa: S106
role="admin",
is_active=True,
)
db_session.add(existing)
await db_session.commit()
response = await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "irrelevant",
"admin_password": "NoSpecial1",
},
)
assert response.status_code == 200
result = response.json()
assert result["auth_enabled"] is True
assert result["admin_created"] is False
class TestAuthLoginAPI:
"""Integration tests for /api/v1/auth/login endpoint."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_login_auth_disabled(self, async_client: AsyncClient):
"""Verify login fails when auth is not enabled."""
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "admin", "password": "password"},
)
assert response.status_code == 400
assert "Authentication is not enabled" in response.json()["detail"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_login_success(self, async_client: AsyncClient):
"""Verify login succeeds with valid credentials after setup."""
# First enable auth
await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "logintest",
"admin_password": "LoginPass1!",
},
)
# Now login
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "logintest", "password": "LoginPass1!"},
)
assert response.status_code == 200
result = response.json()
assert "access_token" in result
assert result["token_type"] == "bearer"
assert result["user"]["username"] == "logintest"
assert result["user"]["role"] == "admin"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_login_invalid_credentials(self, async_client: AsyncClient):
"""Verify login fails with invalid credentials."""
# First enable auth
await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "invalidtest",
"admin_password": "CorrectPass1!",
},
)
# Try login with wrong password
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "invalidtest", "password": "wrongpassword"},
)
assert response.status_code == 401
assert "Incorrect username or password" in response.json()["detail"]
class TestAuthMeAPI:
"""Integration tests for /api/v1/auth/me endpoint."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_me_without_token(self, async_client: AsyncClient):
"""Verify /me fails without authentication token."""
response = await async_client.get("/api/v1/auth/me")
assert response.status_code == 401
@pytest.mark.asyncio
@pytest.mark.integration
async def test_me_with_valid_token(self, async_client: AsyncClient):
"""Verify /me returns user info with valid token."""
# Setup and login
await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "metest",
"admin_password": "MePass1!",
},
)
login_response = await async_client.post(
"/api/v1/auth/login",
json={"username": "metest", "password": "MePass1!"},
)
token = login_response.json()["access_token"]
# Get current user
response = await async_client.get(
"/api/v1/auth/me",
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 200
result = response.json()
assert result["username"] == "metest"
assert result["role"] == "admin"
assert result["is_active"] is True
@pytest.mark.asyncio
@pytest.mark.integration
async def test_me_with_api_key_bearer(self, async_client: AsyncClient, db_session):
"""Verify /me returns synthetic admin user when using API key via Bearer token."""
from backend.app.core.auth import generate_api_key
from backend.app.models.api_key import APIKey
# Create an API key directly in the database
full_key, key_hash, key_prefix = generate_api_key()
api_key = APIKey(name="test-kiosk", key_hash=key_hash, key_prefix=key_prefix, enabled=True)
db_session.add(api_key)
await db_session.commit()
# Call /me with the API key as Bearer token
response = await async_client.get(
"/api/v1/auth/me",
headers={"Authorization": f"Bearer {full_key}"},
)
assert response.status_code == 200
result = response.json()
assert result["id"] == 0
assert result["username"].startswith("api-key:")
assert result["role"] == "admin"
assert result["is_admin"] is True
assert result["is_active"] is True
assert len(result["permissions"]) > 0
@pytest.mark.asyncio
@pytest.mark.integration
async def test_me_with_api_key_header(self, async_client: AsyncClient, db_session):
"""Verify /me returns synthetic admin user when using X-API-Key header."""
from backend.app.core.auth import generate_api_key
from backend.app.models.api_key import APIKey
full_key, key_hash, key_prefix = generate_api_key()
api_key = APIKey(name="test-kiosk-header", key_hash=key_hash, key_prefix=key_prefix, enabled=True)
db_session.add(api_key)
await db_session.commit()
response = await async_client.get(
"/api/v1/auth/me",
headers={"X-API-Key": full_key},
)
assert response.status_code == 200
result = response.json()
assert result["id"] == 0
assert result["username"].startswith("api-key:")
assert result["is_admin"] is True
@pytest.mark.asyncio
@pytest.mark.integration
async def test_me_with_invalid_api_key(self, async_client: AsyncClient):
"""Verify /me rejects invalid API key."""
response = await async_client.get(
"/api/v1/auth/me",
headers={"Authorization": "Bearer bb_invalid_key_value"},
)
assert response.status_code == 401
class TestUsersAPI:
"""Integration tests for /api/v1/users/ endpoints."""
@pytest.fixture
async def auth_token(self, async_client: AsyncClient):
"""Setup auth and return admin token."""
await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "usersadmin",
"admin_password": "AdminPass1!",
},
)
login_response = await async_client.post(
"/api/v1/auth/login",
json={"username": "usersadmin", "password": "AdminPass1!"},
)
return login_response.json()["access_token"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_users_requires_auth(self, async_client: AsyncClient):
"""Verify listing users requires authentication when auth is enabled."""
# First enable auth
await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "authreqadmin",
"admin_password": "AdminPass1!",
},
)
# Now try to list users without a token
response = await async_client.get("/api/v1/users/")
assert response.status_code == 401
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_users_as_admin(self, async_client: AsyncClient, auth_token: str):
"""Verify admin can list users."""
response = await async_client.get(
"/api/v1/users/",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert response.status_code == 200
result = response.json()
assert isinstance(result, list)
assert len(result) >= 1 # At least the admin user
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_user(self, async_client: AsyncClient, auth_token: str):
"""Verify admin can create a new user."""
response = await async_client.post(
"/api/v1/users/",
headers={"Authorization": f"Bearer {auth_token}"},
json={
"username": "newuser",
"password": "Newuserpass1!",
"role": "user",
},
)
assert response.status_code == 201
result = response.json()
assert result["username"] == "newuser"
assert result["role"] == "user"
assert result["is_active"] is True
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_user_duplicate_username(self, async_client: AsyncClient, auth_token: str):
"""Verify creating user with duplicate username fails."""
# Create first user
await async_client.post(
"/api/v1/users/",
headers={"Authorization": f"Bearer {auth_token}"},
json={
"username": "duplicateuser",
"password": "Password123!",
"role": "user",
},
)
# Try to create duplicate
response = await async_client.post(
"/api/v1/users/",
headers={"Authorization": f"Bearer {auth_token}"},
json={
"username": "duplicateuser",
"password": "Password456!",
"role": "user",
},
)
assert response.status_code == 400
assert "Username already exists" in response.json()["detail"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_user(self, async_client: AsyncClient, auth_token: str):
"""Verify admin can update a user."""
# Create user
create_response = await async_client.post(
"/api/v1/users/",
headers={"Authorization": f"Bearer {auth_token}"},
json={
"username": "updateuser",
"password": "Password123!",
"role": "user",
},
)
user_id = create_response.json()["id"]
# Update user
response = await async_client.patch(
f"/api/v1/users/{user_id}",
headers={"Authorization": f"Bearer {auth_token}"},
json={"role": "admin"},
)
assert response.status_code == 200
assert response.json()["role"] == "admin"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_user(self, async_client: AsyncClient, auth_token: str):
"""Verify admin can delete a user."""
# Create user
create_response = await async_client.post(
"/api/v1/users/",
headers={"Authorization": f"Bearer {auth_token}"},
json={
"username": "deleteuser",
"password": "Password123!",
"role": "user",
},
)
user_id = create_response.json()["id"]
# Delete user
response = await async_client.delete(
f"/api/v1/users/{user_id}",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert response.status_code == 204
class TestAuthDisableAPI:
"""Integration tests for /api/v1/auth/disable endpoint."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_disable_auth(self, async_client: AsyncClient):
"""Verify admin can disable authentication."""
# Setup auth
await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "disableadmin",
"admin_password": "AdminPass1!",
},
)
# Login to get token
login_response = await async_client.post(
"/api/v1/auth/login",
json={"username": "disableadmin", "password": "AdminPass1!"},
)
token = login_response.json()["access_token"]
# Disable auth
response = await async_client.post(
"/api/v1/auth/disable",
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 200
assert response.json()["auth_enabled"] is False
# Verify auth is now disabled
status_response = await async_client.get("/api/v1/auth/status")
assert status_response.json()["auth_enabled"] is False
class TestGroupsAPI:
"""Integration tests for /api/v1/groups/ endpoints."""
@pytest.fixture
async def auth_token(self, async_client: AsyncClient):
"""Setup auth and return admin token."""
await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "groupsadmin",
"admin_password": "AdminPass1!",
},
)
login_response = await async_client.post(
"/api/v1/auth/login",
json={"username": "groupsadmin", "password": "AdminPass1!"},
)
return login_response.json()["access_token"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_groups(self, async_client: AsyncClient, auth_token: str):
"""Verify listing groups returns default groups."""
response = await async_client.get(
"/api/v1/groups/",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert response.status_code == 200
groups = response.json()
assert isinstance(groups, list)
# Should have default groups: Administrators, Operators, Viewers
group_names = [g["name"] for g in groups]
assert "Administrators" in group_names
assert "Operators" in group_names
assert "Viewers" in group_names
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_permissions(self, async_client: AsyncClient, auth_token: str):
"""Verify getting available permissions."""
response = await async_client.get(
"/api/v1/groups/permissions",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert response.status_code == 200
permissions = response.json()
assert isinstance(permissions, dict)
# Should have permission categories
assert "Printers" in permissions or len(permissions) > 0
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_group(self, async_client: AsyncClient, auth_token: str):
"""Verify creating a new group."""
response = await async_client.post(
"/api/v1/groups/",
headers={"Authorization": f"Bearer {auth_token}"},
json={
"name": "Custom Group",
"description": "A custom test group",
"permissions": ["printers:read", "archives:read"],
},
)
assert response.status_code == 201
group = response.json()
assert group["name"] == "Custom Group"
assert group["description"] == "A custom test group"
assert "printers:read" in group["permissions"]
assert group["is_system"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_group(self, async_client: AsyncClient, auth_token: str):
"""Verify updating a group."""
# Create a group first
create_response = await async_client.post(
"/api/v1/groups/",
headers={"Authorization": f"Bearer {auth_token}"},
json={
"name": "Update Test Group",
"permissions": ["printers:read"],
},
)
group_id = create_response.json()["id"]
# Update the group
response = await async_client.patch(
f"/api/v1/groups/{group_id}",
headers={"Authorization": f"Bearer {auth_token}"},
json={
"description": "Updated description",
"permissions": ["printers:read", "printers:control"],
},
)
assert response.status_code == 200
group = response.json()
assert group["description"] == "Updated description"
assert "printers:control" in group["permissions"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_cannot_delete_system_group(self, async_client: AsyncClient, auth_token: str):
"""Verify system groups cannot be deleted."""
# Get the Administrators group
list_response = await async_client.get(
"/api/v1/groups/",
headers={"Authorization": f"Bearer {auth_token}"},
)
admin_group = next(g for g in list_response.json() if g["name"] == "Administrators")
# Try to delete it
response = await async_client.delete(
f"/api/v1/groups/{admin_group['id']}",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert response.status_code == 400
assert "system group" in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_custom_group(self, async_client: AsyncClient, auth_token: str):
"""Verify custom groups can be deleted."""
# Create a group
create_response = await async_client.post(
"/api/v1/groups/",
headers={"Authorization": f"Bearer {auth_token}"},
json={"name": "Delete Test Group"},
)
group_id = create_response.json()["id"]
# Delete it
response = await async_client.delete(
f"/api/v1/groups/{group_id}",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert response.status_code == 204
class TestUserGroupsAPI:
"""Integration tests for user-group assignments."""
@pytest.fixture
async def auth_token(self, async_client: AsyncClient):
"""Setup auth and return admin token."""
await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "usergroupadmin",
"admin_password": "AdminPass1!",
},
)
login_response = await async_client.post(
"/api/v1/auth/login",
json={"username": "usergroupadmin", "password": "AdminPass1!"},
)
return login_response.json()["access_token"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_user_with_groups(self, async_client: AsyncClient, auth_token: str):
"""Verify creating a user with group assignments."""
# Get Operators group ID
groups_response = await async_client.get(
"/api/v1/groups/",
headers={"Authorization": f"Bearer {auth_token}"},
)
operators_group = next(g for g in groups_response.json() if g["name"] == "Operators")
# Create user with group
response = await async_client.post(
"/api/v1/users/",
headers={"Authorization": f"Bearer {auth_token}"},
json={
"username": "groupuser",
"password": "Password123!",
"group_ids": [operators_group["id"]],
},
)
assert response.status_code == 201
user = response.json()
assert any(g["name"] == "Operators" for g in user["groups"])
@pytest.mark.asyncio
@pytest.mark.integration
async def test_add_user_to_group(self, async_client: AsyncClient, auth_token: str):
"""Verify adding a user to a group."""
# Create a user
user_response = await async_client.post(
"/api/v1/users/",
headers={"Authorization": f"Bearer {auth_token}"},
json={"username": "addtogroup", "password": "Password123!"},
)
user_id = user_response.json()["id"]
# Get Viewers group
groups_response = await async_client.get(
"/api/v1/groups/",
headers={"Authorization": f"Bearer {auth_token}"},
)
viewers_group = next(g for g in groups_response.json() if g["name"] == "Viewers")
# Add user to group
response = await async_client.post(
f"/api/v1/groups/{viewers_group['id']}/users/{user_id}",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert response.status_code == 204
# Verify user is in group
user_check = await async_client.get(
f"/api/v1/users/{user_id}",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert any(g["name"] == "Viewers" for g in user_check.json()["groups"])
class TestChangePasswordAPI:
"""Integration tests for /api/v1/users/me/change-password endpoint."""
@pytest.fixture
async def user_token(self, async_client: AsyncClient):
"""Setup auth and return regular user token."""
# Enable auth with admin
await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "pwchangeadmin",
"admin_password": "AdminPass1!",
},
)
admin_login = await async_client.post(
"/api/v1/auth/login",
json={"username": "pwchangeadmin", "password": "AdminPass1!"},
)
admin_token = admin_login.json()["access_token"]
# Create a regular user
await async_client.post(
"/api/v1/users/",
headers={"Authorization": f"Bearer {admin_token}"},
json={"username": "pwchangeuser", "password": "Oldpassword123!"},
)
# Login as regular user
user_login = await async_client.post(
"/api/v1/auth/login",
json={"username": "pwchangeuser", "password": "Oldpassword123!"},
)
return user_login.json()["access_token"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_change_password_success(self, async_client: AsyncClient, user_token: str):
"""Verify user can change their own password."""
response = await async_client.post(
"/api/v1/users/me/change-password",
headers={"Authorization": f"Bearer {user_token}"},
json={
"current_password": "Oldpassword123!",
"new_password": "Newpassword456!",
},
)
assert response.status_code == 200
assert "success" in response.json()["message"].lower()
# Verify can login with new password
login_response = await async_client.post(
"/api/v1/auth/login",
json={"username": "pwchangeuser", "password": "Newpassword456!"},
)
assert login_response.status_code == 200
@pytest.mark.asyncio
@pytest.mark.integration
async def test_change_password_wrong_current(self, async_client: AsyncClient, user_token: str):
"""Verify changing password fails with wrong current password."""
response = await async_client.post(
"/api/v1/users/me/change-password",
headers={"Authorization": f"Bearer {user_token}"},
json={
"current_password": "wrongpassword",
"new_password": "Newpassword456!",
},
)
assert response.status_code == 400
assert "incorrect" in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_change_password_requires_auth(self, async_client: AsyncClient):
"""Verify changing password requires authentication."""
response = await async_client.post(
"/api/v1/users/me/change-password",
json={
"current_password": "oldpassword",
"new_password": "Strongpass456!",
},
)
assert response.status_code == 401
class TestAuthMiddlewarePublicRoutes:
"""Tests for auth middleware public route configuration.
These routes must be accessible without authentication, even when auth is enabled,
because browser elements like <img src> and <video src> don't send Authorization headers.
"""
@pytest.fixture
async def enabled_auth(self, async_client: AsyncClient):
"""Enable auth for testing middleware behavior."""
await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "middlewareadmin",
"admin_password": "AdminPass1!",
},
)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_auth_status_is_public(self, async_client: AsyncClient, enabled_auth):
"""Verify /api/v1/auth/status is accessible without auth."""
response = await async_client.get("/api/v1/auth/status")
assert response.status_code == 200
assert "auth_enabled" in response.json()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_auth_login_is_public(self, async_client: AsyncClient, enabled_auth):
"""Verify /api/v1/auth/login is accessible without auth."""
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "middlewareadmin", "password": "AdminPass1!"},
)
# Should not return 401 (unauthorized) - it should either succeed or return
# a different error (like 400 for wrong credentials)
assert response.status_code != 401 or "token" in response.json()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_auth_setup_is_public(self, async_client: AsyncClient):
"""Verify /api/v1/auth/setup is accessible without auth (needed for setup/recovery)."""
# Don't enable auth first - test that setup endpoint itself is accessible
response = await async_client.post(
"/api/v1/auth/setup",
json={"auth_enabled": False},
)
# Should not be 401
assert response.status_code != 401
@pytest.mark.asyncio
@pytest.mark.integration
async def test_updates_version_is_public(self, async_client: AsyncClient, enabled_auth):
"""Verify /api/v1/updates/version is accessible without auth."""
response = await async_client.get("/api/v1/updates/version")
# Should not be 401
assert response.status_code != 401
@pytest.mark.asyncio
@pytest.mark.integration
async def test_protected_route_requires_auth(self, async_client: AsyncClient, enabled_auth):
"""Verify non-public routes return 401 without token."""
response = await async_client.get("/api/v1/printers/")
assert response.status_code == 401
@pytest.mark.asyncio
@pytest.mark.integration
async def test_protected_route_works_with_token(self, async_client: AsyncClient, enabled_auth):
"""Verify non-public routes work with valid token."""
# Login to get token
login_response = await async_client.post(
"/api/v1/auth/login",
json={"username": "middlewareadmin", "password": "AdminPass1!"},
)
token = login_response.json()["access_token"]
# Access protected route
response = await async_client.get(
"/api/v1/printers/",
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 200
@pytest.mark.asyncio
@pytest.mark.integration
async def test_advanced_auth_status_is_public(self, async_client: AsyncClient, enabled_auth):
"""Verify /api/v1/auth/advanced-auth/status is accessible without auth."""
response = await async_client.get("/api/v1/auth/advanced-auth/status")
# Should not be 401 (must be accessible for login page)
assert response.status_code != 401
# Should return valid response (200 with auth status)
if response.status_code == 200:
result = response.json()
assert "advanced_auth_enabled" in result
assert "smtp_configured" in result
@pytest.mark.asyncio
@pytest.mark.integration
async def test_forgot_password_is_public(self, async_client: AsyncClient, enabled_auth):
"""Verify /api/v1/auth/forgot-password is accessible without auth."""
response = await async_client.post(
"/api/v1/auth/forgot-password",
json={"email": "test@example.com"},
)
# Should not be 401 (must be accessible for password reset from login page)
assert response.status_code != 401
# Will likely be 400 (advanced auth not enabled) but that's okay -
# the important thing is it's not blocked by auth middleware
assert response.status_code in [200, 400]
# ===========================================================================
# H-1: Input length validation
# ===========================================================================
class TestInputLengthValidation:
"""LoginRequest and SetupRequest must reject oversized inputs (H-1)."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_login_password_too_long_rejected(self, async_client: AsyncClient):
"""Password exceeding 256 characters must be rejected with 422."""
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "admin", "password": "x" * 257},
)
assert response.status_code == 422
@pytest.mark.asyncio
@pytest.mark.integration
async def test_login_username_too_long_rejected(self, async_client: AsyncClient):
"""Username exceeding 150 characters must be rejected with 422."""
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "u" * 151, "password": "password"},
)
assert response.status_code == 422
@pytest.mark.asyncio
@pytest.mark.integration
async def test_setup_password_too_long_rejected(self, async_client: AsyncClient):
"""SetupRequest admin_password exceeding 256 characters must be rejected with 422."""
response = await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "admin",
"admin_password": "x" * 257,
},
)
assert response.status_code == 422
@pytest.mark.asyncio
@pytest.mark.integration
async def test_login_password_at_limit_accepted(self, async_client: AsyncClient):
"""Password of exactly 256 characters must pass schema validation (may fail auth)."""
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "admin", "password": "x" * 256},
)
# Schema accepts it; auth may reject with 401 (auth disabled) or 400
assert response.status_code != 422