daniel156161
9c731d6e67
Build and Push Docker Container / build-and-push (push) Failing after 51s
- Register quart_common wide-event logging during app setup so every HTTP request emits one canonical structured event. - Replace the inline security middleware with reusable quart_common security middleware wiring and move skip path configuration into app constants. - Add NanoShare-specific wide-event context for health checks, auth/error handlers, file list/edit/delete/serve flows and upload outcomes. - Rename runtime logging/project metadata from simple-picoshare to nanoshare where it is emitted in service context. - Update my_helpers and quart_common submodules for Convex/wide-event integration and reusable security middleware support. - Add NanoShare middleware tests covering safe user context, client IP enrichment, missing Convex handling and Convex security lookup failures.
146 lines
4.6 KiB
Python
146 lines
4.6 KiB
Python
import asyncio
|
|
import importlib
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from quart import Quart, g, session
|
|
|
|
class FakeLogger:
|
|
def __init__(self):
|
|
self.records = []
|
|
|
|
async def info(self, message):
|
|
self.records.append(("info", message))
|
|
|
|
async def warning(self, message):
|
|
self.records.append(("warning", message))
|
|
|
|
async def error(self, message):
|
|
self.records.append(("error", message))
|
|
|
|
class FakeIPBotManager:
|
|
def is_client_ip_always_allowed(self, ip):
|
|
return False
|
|
|
|
class FakeSetupApp:
|
|
def before_request(self, func):
|
|
return func
|
|
|
|
def context_processor(self, func):
|
|
return func
|
|
|
|
class FakeConvex:
|
|
async def is_ip_address_whitelisted_or_blocked(self, ip_address):
|
|
return {"whiteliste": False, "blocked": False}
|
|
|
|
async def is_path_blocked(self, path):
|
|
return False
|
|
|
|
class FailingConvex:
|
|
async def is_ip_address_whitelisted_or_blocked(self, ip_address):
|
|
raise RuntimeError("convex down")
|
|
|
|
def load_middleware(monkeypatch, client_ip="203.0.113.10", logger=None):
|
|
fake_errors = types.ModuleType("routes.handeling.errorsAndBots")
|
|
|
|
async def maybe_a_hacker():
|
|
return "blocked", 418
|
|
|
|
fake_errors.maybe_a_hacker = maybe_a_hacker
|
|
|
|
fake_constens = types.ModuleType("my_modules.app.constens")
|
|
fake_constens.THE_IP_BOT_MANAGER = FakeIPBotManager()
|
|
fake_constens.SKIP_PATH_PREFIXES = ("/static", "/storage")
|
|
fake_constens.SKIP_PATHS = ("/favicon.ico",)
|
|
|
|
fake_logger_module = types.ModuleType("my_modules.app.logger")
|
|
fake_logger_module.logger = logger or FakeLogger()
|
|
|
|
fake_functions = types.ModuleType("my_modules.functions")
|
|
fake_functions.get_ip = lambda: client_ip
|
|
|
|
fake_setup = types.ModuleType("my_modules.app.setup")
|
|
fake_setup.app = FakeSetupApp()
|
|
|
|
monkeypatch.setitem(sys.modules, "routes.handeling.errorsAndBots", fake_errors)
|
|
monkeypatch.setitem(sys.modules, "my_modules.app.constens", fake_constens)
|
|
monkeypatch.setitem(sys.modules, "my_modules.app.logger", fake_logger_module)
|
|
monkeypatch.setitem(sys.modules, "my_modules.functions", fake_functions)
|
|
monkeypatch.setitem(sys.modules, "my_modules.app.setup", fake_setup)
|
|
sys.modules.pop("my_modules.middleware", None)
|
|
|
|
return importlib.import_module("my_modules.middleware")
|
|
|
|
def test_middleware_adds_user_and_client_context(monkeypatch):
|
|
async def run_test():
|
|
middleware = load_middleware(monkeypatch)
|
|
monkeypatch.setattr(middleware, "get_ip", lambda: "203.0.113.10")
|
|
monkeypatch.setattr(middleware, "logger", FakeLogger())
|
|
|
|
app = Quart(__name__)
|
|
app.secret_key = "test-secret"
|
|
app.convex = FakeConvex()
|
|
|
|
async with app.test_request_context("/files"):
|
|
g.wide_event = {}
|
|
session["user"] = {"sub": "user_123", "preferred_username": "demo"}
|
|
session["login_at"] = 1
|
|
|
|
result = await middleware.custom_middleware()
|
|
|
|
assert result is None
|
|
assert g.wide_event["user"]["id"] == "user_123"
|
|
assert g.wide_event["user"]["name"] == "demo"
|
|
assert g.wide_event["user"]["authenticated"] is True
|
|
assert g.wide_event["user"]["session_age_seconds"] >= 0
|
|
assert g.wide_event["client"]["ip"] == "203.0.113.10"
|
|
|
|
asyncio.run(run_test())
|
|
|
|
def test_middleware_marks_missing_convex_as_skipped(monkeypatch):
|
|
async def run_test():
|
|
middleware = load_middleware(monkeypatch, client_ip="203.0.113.11")
|
|
monkeypatch.setattr(middleware, "logger", FakeLogger())
|
|
|
|
app = Quart(__name__)
|
|
app.secret_key = "test-secret"
|
|
|
|
async with app.test_request_context("/files"):
|
|
g.wide_event = {}
|
|
|
|
result = await middleware.custom_middleware()
|
|
|
|
assert result is None
|
|
assert g.wide_event["client"]["ip"] == "203.0.113.11"
|
|
assert g.wide_event["security"] == {
|
|
"convex_missing": True,
|
|
"middleware_skipped": True,
|
|
}
|
|
|
|
asyncio.run(run_test())
|
|
|
|
def test_middleware_records_convex_security_failures(monkeypatch):
|
|
async def run_test():
|
|
fake_logger = FakeLogger()
|
|
middleware = load_middleware(monkeypatch, client_ip="203.0.113.12", logger=fake_logger)
|
|
|
|
app = Quart(__name__)
|
|
app.secret_key = "test-secret"
|
|
app.convex = FailingConvex()
|
|
|
|
async with app.test_request_context("/files"):
|
|
g.wide_event = {}
|
|
|
|
result = await middleware.custom_middleware()
|
|
|
|
assert result is None
|
|
assert g.wide_event["client"]["ip"] == "203.0.113.12"
|
|
assert g.wide_event["security"] == {"ip_lookup_failed": True}
|
|
assert g.wide_event["error"]["type"] == "RuntimeError"
|
|
assert fake_logger.records == [("error", "[MIDDLEWARE] Convex ip_lookup failed: convex down")]
|
|
|
|
asyncio.run(run_test())
|