mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
* 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.
970 lines
35 KiB
Python
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
|