feat(middleware): harden routing with fail-closed policy, custom subdomain management, and perf fixes
Some checks failed
CI / pytest (push) Waiting to run
CI / ruff (push) Successful in 12s
CI / validate (push) Successful in 26s
CI / dependency-scanning (push) Successful in 31s
CI / docs (push) Has been cancelled
CI / deploy (push) Has been cancelled

- Fix IPv6 host parsing with _strip_port() utility
- Remove dangerous StorePlatform→Store.subdomain silent fallback
- Close storefront gate bypass when frontend_type is None
- Add custom subdomain management UI and API for stores
- Add domain health diagnostic tool
- Convert db.add() in loops to db.add_all() (24 PERF-006 fixes)
- Add tests for all new functionality (18 subdomain service tests)
- Add .github templates for validator compliance

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-15 18:13:01 +01:00
parent 07fab01f6a
commit 540205402f
38 changed files with 1827 additions and 134 deletions

View File

@@ -0,0 +1,215 @@
# tests/unit/middleware/test_fail_closed_store_context.py
"""
Unit tests for fail-closed store context resolution.
Verifies that subdomain detection with a platform context:
1. Tries StorePlatform.custom_subdomain first (platform-specific override)
2. Falls back to Store.subdomain but ONLY if the store has an active
membership on the detected platform (prevents cross-tenant leaks)
3. Returns None if the store has no membership on the platform
"""
from unittest.mock import MagicMock, Mock
import pytest
from sqlalchemy.orm import Session
from middleware.store_context import StoreContextManager
@pytest.mark.unit
@pytest.mark.middleware
class TestFailClosedStoreContext:
"""Ensure subdomain+platform blocks cross-tenant resolution."""
def test_subdomain_not_found_at_all_returns_none(self):
"""
Platform + subdomain + no custom_subdomain match + no Store.subdomain match → None.
"""
mock_db = Mock(spec=Session)
mock_platform = Mock()
mock_platform.id = 2
mock_platform.code = "loyalty"
# All queries return None
mock_db.query.return_value.filter.return_value.first.return_value = None
mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = None
context = {
"detection_method": "subdomain",
"subdomain": "nonexistent",
"_platform": mock_platform,
}
store = StoreContextManager.get_store_from_context(mock_db, context)
assert store is None
def test_subdomain_found_with_platform_membership_returns_store(self):
"""
Platform + subdomain + Store.subdomain found + store HAS membership
on this platform → return store.
e.g. wizatech.omsflow.lu where WIZATECH is on OMS platform.
"""
mock_db = Mock(spec=Session)
mock_platform = Mock()
mock_platform.id = 1
mock_platform.code = "oms"
mock_store = Mock()
mock_store.id = 10
mock_store.is_active = True
mock_store.name = "WizaTech"
mock_membership = Mock() # Active StorePlatform entry
call_count = [0]
def side_effect_query(model):
result = MagicMock()
call_count[0] += 1
if call_count[0] == 1:
# StorePlatform.custom_subdomain lookup → no match
result.filter.return_value.first.return_value = None
elif call_count[0] == 2:
# Store.subdomain lookup → found
result.filter.return_value.filter.return_value.first.return_value = mock_store
else:
# StorePlatform membership check → exists
result.filter.return_value.first.return_value = mock_membership
return result
mock_db.query.side_effect = side_effect_query
context = {
"detection_method": "subdomain",
"subdomain": "wizatech",
"_platform": mock_platform,
}
store = StoreContextManager.get_store_from_context(mock_db, context)
assert store is mock_store
def test_subdomain_found_without_platform_membership_returns_none(self):
"""
Platform + subdomain + Store.subdomain found but store has NO
membership on this platform → None (cross-tenant blocked).
e.g. loyalty-only-store.omsflow.lu where the store is only on loyalty.
"""
mock_db = Mock(spec=Session)
mock_platform = Mock()
mock_platform.id = 1
mock_platform.code = "oms"
mock_store = Mock()
mock_store.id = 10
mock_store.is_active = True
mock_store.name = "LoyaltyOnlyStore"
call_count = [0]
def side_effect_query(model):
result = MagicMock()
call_count[0] += 1
if call_count[0] == 1:
# StorePlatform.custom_subdomain lookup → no match
result.filter.return_value.first.return_value = None
elif call_count[0] == 2:
# Store.subdomain lookup → found
result.filter.return_value.filter.return_value.first.return_value = mock_store
else:
# StorePlatform membership check → NOT found (no membership)
result.filter.return_value.first.return_value = None
return result
mock_db.query.side_effect = side_effect_query
context = {
"detection_method": "subdomain",
"subdomain": "loyalty-only",
"_platform": mock_platform,
}
store = StoreContextManager.get_store_from_context(mock_db, context)
assert store is None
def test_custom_subdomain_found_returns_store(self):
"""Platform + subdomain + custom_subdomain found → correct store."""
mock_db = Mock(spec=Session)
mock_platform = Mock()
mock_platform.id = 2
mock_platform.code = "loyalty"
mock_store_platform = Mock()
mock_store_platform.store_id = 42
mock_store = Mock()
mock_store.id = 42
mock_store.is_active = True
mock_store.name = "Acme Rewards"
call_count = [0]
def side_effect_query(model):
result = MagicMock()
call_count[0] += 1
if call_count[0] == 1:
result.filter.return_value.first.return_value = mock_store_platform
else:
result.filter.return_value.first.return_value = mock_store
return result
mock_db.query.side_effect = side_effect_query
context = {
"detection_method": "subdomain",
"subdomain": "acme-rewards",
"_platform": mock_platform,
}
store = StoreContextManager.get_store_from_context(mock_db, context)
assert store is mock_store
def test_no_platform_subdomain_uses_global_store(self):
"""No platform + subdomain → uses Store.subdomain (dev mode)."""
mock_db = Mock(spec=Session)
mock_store = Mock()
mock_store.is_active = True
mock_store.name = "Acme"
mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = mock_store
context = {
"detection_method": "subdomain",
"subdomain": "acme",
# No "_platform" key — dev mode
}
store = StoreContextManager.get_store_from_context(mock_db, context)
assert store is mock_store
def test_path_based_uses_global_store(self):
"""Path-based detection always uses Store.subdomain (unchanged)."""
mock_db = Mock(spec=Session)
mock_platform = Mock()
mock_platform.id = 1
mock_store = Mock()
mock_store.is_active = True
mock_store.name = "Acme"
mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = mock_store
context = {
"detection_method": "path",
"subdomain": "ACME",
"_platform": mock_platform,
}
store = StoreContextManager.get_store_from_context(mock_db, context)
assert store is mock_store

View File

@@ -188,13 +188,13 @@ class TestGetFrontendTypeHelper:
assert result == FrontendType.ADMIN
def test_get_frontend_type_default(self):
"""Test getting frontend type returns PLATFORM as default."""
"""Test getting frontend type returns None when not set (fail-aware)."""
request = Mock(spec=Request)
request.state = Mock(spec=[]) # No frontend_type attribute
result = get_frontend_type(request)
assert result == FrontendType.PLATFORM
assert result is None
def test_get_frontend_type_for_all_types(self):
"""Test getting all frontend types."""

View File

@@ -0,0 +1,63 @@
# tests/unit/middleware/test_ipv6_host_parsing.py
"""
Unit tests for _strip_port() IPv6-safe host parsing utility.
Ensures the middleware correctly strips ports from:
- IPv4 hosts (localhost:8000)
- IPv6 hosts ([::1]:8000)
- Bare hostnames (example.com)
- Edge cases (empty, malformed brackets)
"""
import pytest
from middleware.platform_context import _strip_port
@pytest.mark.unit
@pytest.mark.middleware
class TestStripPort:
"""Test _strip_port() handles all host formats correctly."""
def test_ipv4_with_port(self):
assert _strip_port("127.0.0.1:8000") == "127.0.0.1"
def test_ipv4_without_port(self):
assert _strip_port("127.0.0.1") == "127.0.0.1"
def test_localhost_with_port(self):
assert _strip_port("localhost:9999") == "localhost"
def test_localhost_without_port(self):
assert _strip_port("localhost") == "localhost"
def test_domain_with_port(self):
assert _strip_port("example.com:443") == "example.com"
def test_domain_without_port(self):
assert _strip_port("example.com") == "example.com"
def test_ipv6_with_brackets_and_port(self):
assert _strip_port("[::1]:8000") == "::1"
def test_ipv6_with_brackets_no_port(self):
assert _strip_port("[::1]") == "::1"
def test_ipv6_full_address_with_port(self):
assert _strip_port("[2001:db8::1]:443") == "2001:db8::1"
def test_ipv6_full_address_no_port(self):
assert _strip_port("[2001:db8::1]") == "2001:db8::1"
def test_empty_string(self):
assert _strip_port("") == ""
def test_bare_hostname(self):
assert _strip_port("myhost") == "myhost"
def test_subdomain_with_port(self):
assert _strip_port("store.omsflow.lu:8080") == "store.omsflow.lu"
def test_malformed_brackets_no_closing(self):
"""Malformed bracket with no closing ] returns as-is."""
assert _strip_port("[::1") == "[::1"

View File

@@ -63,34 +63,15 @@ class TestCustomSubdomainResolution:
assert store is mock_store
def test_custom_subdomain_not_found_falls_back_to_store_subdomain(self):
"""When custom_subdomain doesn't match, fall back to Store.subdomain."""
def test_custom_subdomain_not_found_returns_none_on_platform(self):
"""When custom_subdomain doesn't match on a platform, return None (fail-closed)."""
mock_db = Mock(spec=Session)
mock_platform = Mock()
mock_platform.id = 2
mock_platform.code = "loyalty"
mock_store = Mock()
mock_store.is_active = True
mock_store.name = "Acme Corp"
# Query sequence:
# 1. StorePlatform query → None (no custom_subdomain match)
# 2. Store query → mock_store (subdomain match)
call_count = [0]
def side_effect_query(model):
result = MagicMock()
call_count[0] += 1
if call_count[0] == 1:
# StorePlatform query → no match
result.filter.return_value.first.return_value = None
else:
# Store.subdomain fallback
result.filter.return_value.filter.return_value.first.return_value = mock_store
return result
mock_db.query.side_effect = side_effect_query
# StorePlatform query → no match
mock_db.query.return_value.filter.return_value.first.return_value = None
context = {
"detection_method": "subdomain",
@@ -100,6 +81,25 @@ class TestCustomSubdomainResolution:
store = StoreContextManager.get_store_from_context(mock_db, context)
assert store is None
def test_no_platform_subdomain_uses_global_store_subdomain(self):
"""When no platform context, subdomain detection uses global Store.subdomain."""
mock_db = Mock(spec=Session)
mock_store = Mock()
mock_store.is_active = True
mock_store.name = "Acme Corp"
mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = mock_store
context = {
"detection_method": "subdomain",
"subdomain": "acme",
# No "_platform" key — dev mode
}
store = StoreContextManager.get_store_from_context(mock_db, context)
assert store is mock_store
def test_no_platform_skips_custom_subdomain_lookup(self):

View File

@@ -172,10 +172,51 @@ class TestStorefrontAccessMiddlewarePassthrough:
call_next.assert_called_once_with(request)
@pytest.mark.asyncio
async def test_no_frontend_type_passes_through(self):
"""Test request with no frontend_type set passes through."""
async def test_no_frontend_type_non_storefront_passes_through(self):
"""Test request with no frontend_type on non-storefront path passes through."""
middleware = StorefrontAccessMiddleware(app=None)
request = _make_request()
request = _make_request(path="/admin/dashboard")
request.state.frontend_type = None
call_next = AsyncMock(return_value=Mock())
await middleware.dispatch(request, call_next)
call_next.assert_called_once_with(request)
@pytest.mark.asyncio
async def test_no_frontend_type_storefront_path_blocked(self):
"""Test request with no frontend_type on storefront path is blocked (fail-closed)."""
middleware = StorefrontAccessMiddleware(app=None)
request = _make_request(path="/storefront/products")
request.state.frontend_type = None
request.state.store = None
call_next = AsyncMock()
with patch("app.templates_config.templates") as mock_templates:
mock_templates.TemplateResponse.return_value = Mock(status_code=403)
await middleware.dispatch(request, call_next)
call_next.assert_not_called()
@pytest.mark.asyncio
async def test_no_frontend_type_storefront_api_blocked_json(self):
"""Test request with no frontend_type on storefront API returns JSON 403."""
middleware = StorefrontAccessMiddleware(app=None)
request = _make_request(path="/api/v1/storefront/cart")
request.state.frontend_type = None
call_next = AsyncMock()
response = await middleware.dispatch(request, call_next)
call_next.assert_not_called()
assert isinstance(response, JSONResponse)
assert response.status_code == 403
@pytest.mark.asyncio
async def test_no_frontend_type_static_passes_through(self):
"""Test request with no frontend_type on static path passes through."""
middleware = StorefrontAccessMiddleware(app=None)
request = _make_request(path="/static/css/style.css")
request.state.frontend_type = None
call_next = AsyncMock(return_value=Mock())

View File

@@ -0,0 +1,126 @@
# tests/unit/middleware/test_storefront_gate_bypass.py
"""
Unit tests for the storefront gate bypass safety net.
Ensures that when frontend_type is None (upstream middleware failed),
storefront paths are BLOCKED rather than passed through. Non-storefront
paths with None frontend_type should pass through unchanged.
"""
from unittest.mock import AsyncMock, Mock, patch
import pytest
from fastapi import Request
from starlette.responses import JSONResponse
from middleware.storefront_access import (
StorefrontAccessMiddleware,
_looks_like_storefront,
)
@pytest.mark.unit
@pytest.mark.middleware
class TestLooksLikeStorefront:
"""Test the _looks_like_storefront helper."""
def test_storefront_page_path(self):
assert _looks_like_storefront("/storefront/products") is True
def test_storefront_api_path(self):
assert _looks_like_storefront("/api/v1/storefront/cart") is True
def test_admin_path(self):
assert _looks_like_storefront("/admin/dashboard") is False
def test_static_path(self):
assert _looks_like_storefront("/static/css/style.css") is False
def test_store_path(self):
assert _looks_like_storefront("/store/ACME/login") is False
def test_root_path(self):
assert _looks_like_storefront("/") is False
@pytest.mark.unit
@pytest.mark.middleware
class TestStorefrontGateBypass:
"""Test that None frontend_type can't bypass the storefront gate."""
@pytest.mark.asyncio
async def test_none_frontend_type_storefront_page_blocked(self):
"""frontend_type=None + /storefront/ path → blocked with HTML 403."""
middleware = StorefrontAccessMiddleware(app=None)
request = Mock(spec=Request)
request.url = Mock(path="/storefront/products")
request.state = Mock()
request.state.frontend_type = None
request.state.store = None
request.state.language = "en"
request.state.theme = None
call_next = AsyncMock()
with patch("app.templates_config.templates") as mock_templates:
mock_templates.TemplateResponse.return_value = Mock(status_code=403)
await middleware.dispatch(request, call_next)
call_next.assert_not_called()
@pytest.mark.asyncio
async def test_none_frontend_type_storefront_api_blocked(self):
"""frontend_type=None + /api/v1/storefront/ path → JSON 403."""
middleware = StorefrontAccessMiddleware(app=None)
request = Mock(spec=Request)
request.url = Mock(path="/api/v1/storefront/cart")
request.state = Mock()
request.state.frontend_type = None
call_next = AsyncMock()
response = await middleware.dispatch(request, call_next)
call_next.assert_not_called()
assert isinstance(response, JSONResponse)
assert response.status_code == 403
@pytest.mark.asyncio
async def test_none_frontend_type_admin_passes_through(self):
"""frontend_type=None + /admin/ path → passes through (not storefront)."""
middleware = StorefrontAccessMiddleware(app=None)
request = Mock(spec=Request)
request.url = Mock(path="/admin/dashboard")
request.state = Mock()
request.state.frontend_type = None
call_next = AsyncMock(return_value=Mock())
await middleware.dispatch(request, call_next)
call_next.assert_called_once_with(request)
@pytest.mark.asyncio
async def test_none_frontend_type_static_passes_through(self):
"""frontend_type=None + /static/ path → passes through."""
middleware = StorefrontAccessMiddleware(app=None)
request = Mock(spec=Request)
request.url = Mock(path="/static/css/style.css")
request.state = Mock()
request.state.frontend_type = None
call_next = AsyncMock(return_value=Mock())
await middleware.dispatch(request, call_next)
call_next.assert_called_once_with(request)
@pytest.mark.asyncio
async def test_none_frontend_type_root_passes_through(self):
"""frontend_type=None + / path → passes through."""
middleware = StorefrontAccessMiddleware(app=None)
request = Mock(spec=Request)
request.url = Mock(path="/")
request.state = Mock()
request.state.frontend_type = None
call_next = AsyncMock(return_value=Mock())
await middleware.dispatch(request, call_next)
call_next.assert_called_once_with(request)