256 lines
13 KiB
Python
256 lines
13 KiB
Python
|
"""Isolated ASGI tests: no Docker, database, listening socket, or test dependency."""
|
||
|
|
import asyncio
|
||
|
|
import hashlib
|
||
|
|
from http.cookies import SimpleCookie
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
import time
|
||
|
|
import unittest
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
from fastapi import FastAPI
|
||
|
|
from starlette.responses import HTMLResponse
|
||
|
|
|
||
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
|
|
from service import web_auth
|
||
|
|
|
||
|
|
|
||
|
|
PASSWORD = "isolated-test-password"
|
||
|
|
SALT = bytes(range(16))
|
||
|
|
PASSWORD_HASH = "scrypt$16384$8$1$" + SALT.hex() + "$" + hashlib.scrypt(PASSWORD.encode(), salt=SALT, n=16384, r=8, p=1, dklen=32).hex()
|
||
|
|
ENV = {"AUTH_ENABLED": "true", "WOLIN_AUTH_USER": "admin", "WOLIN_AUTH_PASSWORD_HASH": PASSWORD_HASH,
|
||
|
|
"WOLIN_SESSION_SECRET": "isolated-test-secret-with-32-or-more-characters", "AUTH_COOKIE_SECURE": "false"}
|
||
|
|
|
||
|
|
|
||
|
|
def make_app(overrides=None, removed=()):
|
||
|
|
app = FastAPI()
|
||
|
|
|
||
|
|
@app.get("/")
|
||
|
|
async def home():
|
||
|
|
return HTMLResponse("protected application")
|
||
|
|
|
||
|
|
@app.get("/api/records")
|
||
|
|
async def records():
|
||
|
|
return {"records": [1, 2]}
|
||
|
|
|
||
|
|
@app.api_route("/api/records", methods=["POST", "PUT", "PATCH", "DELETE"])
|
||
|
|
async def modify_records():
|
||
|
|
return {"success": True}
|
||
|
|
|
||
|
|
@app.get("/healthz")
|
||
|
|
async def health():
|
||
|
|
return {"status": "ok"}
|
||
|
|
|
||
|
|
environment = {**ENV, **(overrides or {})}
|
||
|
|
for key in removed:
|
||
|
|
environment.pop(key, None)
|
||
|
|
with patch.dict(os.environ, environment, clear=True):
|
||
|
|
web_auth.install_web_auth(app)
|
||
|
|
return app
|
||
|
|
|
||
|
|
|
||
|
|
async def request(app, path="/", method="GET", data=None, headers=None, root="", prefixed=True, client="127.0.0.1", raw=None):
|
||
|
|
body = raw if raw is not None else (json.dumps(data).encode() if data is not None else b"")
|
||
|
|
request_headers = {"host": "testserver", **({"content-type": "application/json"} if data is not None else {}), **(headers or {})}
|
||
|
|
actual_path = root + path if prefixed else path
|
||
|
|
scope = {"type": "http", "asgi": {"version": "3.0"}, "http_version": "1.1", "method": method,
|
||
|
|
"scheme": "http", "path": actual_path, "raw_path": actual_path.encode(), "query_string": b"",
|
||
|
|
"root_path": root, "headers": [(k.lower().encode(), v.encode()) for k, v in request_headers.items()],
|
||
|
|
"client": (client, 50000), "server": ("testserver", 80)}
|
||
|
|
received = False
|
||
|
|
result = {"status": None, "headers": {}, "body": b""}
|
||
|
|
|
||
|
|
async def receive():
|
||
|
|
nonlocal received
|
||
|
|
if not received:
|
||
|
|
received = True
|
||
|
|
return {"type": "http.request", "body": body, "more_body": False}
|
||
|
|
await asyncio.Event().wait()
|
||
|
|
|
||
|
|
async def send(message):
|
||
|
|
if message["type"] == "http.response.start":
|
||
|
|
result["status"] = message["status"]
|
||
|
|
result["headers"] = {k.decode().lower(): v.decode() for k, v in message["headers"]}
|
||
|
|
elif message["type"] == "http.response.body":
|
||
|
|
result["body"] += message.get("body", b"")
|
||
|
|
|
||
|
|
await app(scope, receive, send)
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
class WebAuthTests(unittest.IsolatedAsyncioTestCase):
|
||
|
|
async def asyncSetUp(self):
|
||
|
|
self.app = make_app()
|
||
|
|
self.temp = tempfile.TemporaryDirectory()
|
||
|
|
page = Path(self.temp.name) / "login.html"
|
||
|
|
page.write_text("<html>form login</html>", encoding="utf-8")
|
||
|
|
self.page_patch = patch.object(web_auth, "LOGIN_HTML", page)
|
||
|
|
self.page_patch.start()
|
||
|
|
|
||
|
|
async def asyncTearDown(self):
|
||
|
|
self.page_patch.stop()
|
||
|
|
self.temp.cleanup()
|
||
|
|
|
||
|
|
async def login(self, app=None, **kwargs):
|
||
|
|
response = await request(app or self.app, "/auth/login", "POST", {"username": "admin", "password": PASSWORD}, **kwargs)
|
||
|
|
self.assertEqual(response["status"], 200, response)
|
||
|
|
self.assertEqual(json.loads(response["body"]), {"success": True})
|
||
|
|
return response
|
||
|
|
|
||
|
|
def cookie(self, response):
|
||
|
|
parsed = SimpleCookie()
|
||
|
|
parsed.load(response["headers"]["set-cookie"])
|
||
|
|
return web_auth.COOKIE_NAME + "=" + parsed[web_auth.COOKIE_NAME].value
|
||
|
|
|
||
|
|
async def test_success_and_cookie_flags(self):
|
||
|
|
response = await self.login()
|
||
|
|
header = response["headers"]["set-cookie"]
|
||
|
|
for flag in ["HttpOnly", "SameSite=strict", "Max-Age=28800", "Path=/"]:
|
||
|
|
self.assertIn(flag, header)
|
||
|
|
self.assertNotIn("Secure", header)
|
||
|
|
result = await request(self.app, "/api/records", headers={"cookie": self.cookie(response)})
|
||
|
|
self.assertEqual(result["status"], 200)
|
||
|
|
self.assertEqual(json.loads(result["body"]), {"records": [1, 2]})
|
||
|
|
|
||
|
|
async def test_secure_by_default(self):
|
||
|
|
response = await self.login(make_app(removed=("AUTH_COOKIE_SECURE",)))
|
||
|
|
self.assertIn("Secure", response["headers"]["set-cookie"])
|
||
|
|
|
||
|
|
async def test_wrong_password_and_user_have_generic_error(self):
|
||
|
|
a = await request(self.app, "/auth/login", "POST", {"username": "admin", "password": "wrong"})
|
||
|
|
b = await request(self.app, "/auth/login", "POST", {"username": "unknown", "password": PASSWORD})
|
||
|
|
self.assertEqual(a["status"], 401)
|
||
|
|
self.assertEqual(a["body"], b["body"])
|
||
|
|
self.assertNotIn("www-authenticate", a["headers"])
|
||
|
|
self.assertNotIn("set-cookie", a["headers"])
|
||
|
|
|
||
|
|
async def test_tampered_and_malformed_session_rejected(self):
|
||
|
|
cookie = self.cookie(await self.login())
|
||
|
|
payload, signature = cookie.split(".")
|
||
|
|
tampered = payload + "." + ("A" if signature[0] != "A" else "B") + signature[1:]
|
||
|
|
for value in [tampered, "wolin_session=invalid", "wolin_session=" + "a" * 2048]:
|
||
|
|
result = await request(self.app, "/api/records", headers={"cookie": value})
|
||
|
|
self.assertEqual(result["status"], 401)
|
||
|
|
|
||
|
|
async def test_expired_session_rejected(self):
|
||
|
|
cookie = self.cookie(await self.login())
|
||
|
|
future = time.time() + web_auth.SESSION_SECONDS + 1
|
||
|
|
with patch.object(web_auth.time, "time", return_value=future):
|
||
|
|
result = await request(self.app, "/api/records", headers={"cookie": cookie})
|
||
|
|
self.assertEqual(result["status"], 401)
|
||
|
|
|
||
|
|
async def test_root_path_cookie_redirect_and_routes(self):
|
||
|
|
response = await self.login(root="/wolin")
|
||
|
|
self.assertIn("Path=/wolin/", response["headers"]["set-cookie"])
|
||
|
|
result = await request(self.app, "/api/records", root="/wolin", headers={"cookie": self.cookie(response)})
|
||
|
|
self.assertEqual(result["status"], 200)
|
||
|
|
for prefixed in [True, False]:
|
||
|
|
result = await request(self.app, "/", root="/wolin", prefixed=prefixed, headers={"accept": "text/html"})
|
||
|
|
self.assertEqual(result["status"], 303)
|
||
|
|
self.assertEqual(result["headers"]["location"], "/wolin/login")
|
||
|
|
page = await request(self.app, "/login", root="/wolin", prefixed=prefixed)
|
||
|
|
self.assertEqual(page["status"], 200)
|
||
|
|
self.assertIn(b"form login", page["body"])
|
||
|
|
|
||
|
|
async def test_all_protected_api_docs_and_static_paths_return_401(self):
|
||
|
|
for path in ["/api/records", "/api", "/docs", "/docs/oauth2-redirect", "/redoc", "/openapi.json", "/static/app.js"]:
|
||
|
|
for root in ["", "/wolin"]:
|
||
|
|
result = await request(self.app, path, root=root, headers={"accept": "text/html"})
|
||
|
|
self.assertEqual(result["status"], 401, (path, result))
|
||
|
|
self.assertNotIn("www-authenticate", result["headers"])
|
||
|
|
self.assertEqual(result["headers"]["content-type"], "application/json")
|
||
|
|
# An API request outside the prefix is protected as well.
|
||
|
|
result = await request(self.app, "/api/records", root="/wolin", prefixed=False)
|
||
|
|
self.assertEqual(result["status"], 401)
|
||
|
|
|
||
|
|
async def test_cross_site_login_and_logout_rejected(self):
|
||
|
|
for path in ["/auth/login", "/auth/logout"]:
|
||
|
|
for headers in [{"origin": "https://evil.example"}, {"origin": "null"}, {"origin": "http://testserver.evil.example"}, {"sec-fetch-site": "cross-site"}, {"origin": "http://testserver:bad"}]:
|
||
|
|
result = await request(self.app, path, "POST", {"username": "admin", "password": PASSWORD}, headers=headers)
|
||
|
|
self.assertEqual(result["status"], 403, (headers, result))
|
||
|
|
self.assertNotIn("set-cookie", result["headers"])
|
||
|
|
await self.login(headers={"origin": "http://testserver", "sec-fetch-site": "same-origin"})
|
||
|
|
|
||
|
|
async def test_logout_clears_scoped_cookie(self):
|
||
|
|
cookie = self.cookie(await self.login(root="/wolin"))
|
||
|
|
response = await request(self.app, "/auth/logout", "POST", root="/wolin", headers={"cookie": cookie, "origin": "http://testserver"})
|
||
|
|
self.assertEqual(response["status"], 200)
|
||
|
|
for flag in ["Max-Age=0", "Path=/wolin/", "HttpOnly", "SameSite=strict"]:
|
||
|
|
self.assertIn(flag, response["headers"]["set-cookie"])
|
||
|
|
result = await request(self.app, "/api/records", root="/wolin", headers={"cookie": self.cookie(response)})
|
||
|
|
self.assertEqual(result["status"], 401)
|
||
|
|
|
||
|
|
async def test_cross_origin_authenticated_writes_are_rejected(self):
|
||
|
|
cookie = self.cookie(await self.login())
|
||
|
|
for method in ["POST", "PUT", "PATCH", "DELETE"]:
|
||
|
|
for origin in ["http://testserver:8004", "https://evil.example"]:
|
||
|
|
result = await request(self.app, "/api/records", method, headers={"cookie": cookie, "origin": origin})
|
||
|
|
self.assertEqual(result["status"], 403)
|
||
|
|
result = await request(self.app, "/api/records", method, headers={"cookie": cookie, "origin": "http://testserver"})
|
||
|
|
self.assertEqual(result["status"], 200)
|
||
|
|
result = await request(self.app, "/api/records", "POST", headers={"cookie": cookie, "sec-fetch-site": "cross-site"})
|
||
|
|
self.assertEqual(result["status"], 403)
|
||
|
|
result = await request(self.app, "/api/records", "POST", headers={"origin": "https://evil.example"})
|
||
|
|
self.assertEqual(result["status"], 401)
|
||
|
|
|
||
|
|
async def test_health_is_public_and_get_only(self):
|
||
|
|
self.assertEqual((await request(self.app, "/healthz"))["status"], 200)
|
||
|
|
self.assertEqual((await request(self.app, "/healthz", "POST"))["status"], 401)
|
||
|
|
|
||
|
|
async def test_auth_disabled_keeps_original_local_behavior(self):
|
||
|
|
for app in [make_app({"AUTH_ENABLED": "false"}, removed=("WOLIN_AUTH_PASSWORD_HASH", "WOLIN_SESSION_SECRET")), make_app(removed=("AUTH_ENABLED",))]:
|
||
|
|
self.assertEqual((await request(app, "/api/records"))["status"], 200)
|
||
|
|
self.assertEqual((await request(app, "/login"))["status"], 404)
|
||
|
|
|
||
|
|
async def test_rate_limit_five_failures_per_minute(self):
|
||
|
|
for _ in range(5):
|
||
|
|
result = await request(self.app, "/auth/login", "POST", {"username": "admin", "password": "wrong"})
|
||
|
|
self.assertEqual(result["status"], 401)
|
||
|
|
result = await request(self.app, "/auth/login", "POST", {"username": "admin", "password": PASSWORD})
|
||
|
|
self.assertEqual(result["status"], 429)
|
||
|
|
self.assertIn("retry-after", result["headers"])
|
||
|
|
await self.login(client="127.0.0.2")
|
||
|
|
future = time.monotonic() + 61
|
||
|
|
with patch.object(web_auth, "monotonic", return_value=future):
|
||
|
|
await self.login()
|
||
|
|
|
||
|
|
async def test_success_resets_failure_count(self):
|
||
|
|
for _ in range(4):
|
||
|
|
await request(self.app, "/auth/login", "POST", {"username": "admin", "password": "wrong"})
|
||
|
|
await self.login()
|
||
|
|
result = await request(self.app, "/auth/login", "POST", {"username": "admin", "password": "wrong"})
|
||
|
|
self.assertEqual(result["status"], 401)
|
||
|
|
|
||
|
|
async def test_malformed_and_oversized_json(self):
|
||
|
|
for raw, status in [(b"[1,2]", 400), (b"not json", 400), (b"{}", 400), (b"x" * 4097, 413)]:
|
||
|
|
result = await request(self.app, "/auth/login", "POST", raw=raw, headers={"content-type": "application/json"})
|
||
|
|
self.assertEqual(result["status"], status)
|
||
|
|
self.assertNotIn("set-cookie", result["headers"])
|
||
|
|
result = await request(self.app, "/auth/login", "POST", raw=b"user=admin")
|
||
|
|
self.assertEqual(result["status"], 415)
|
||
|
|
|
||
|
|
|
||
|
|
class ConfigurationTests(unittest.TestCase):
|
||
|
|
def test_missing_or_invalid_configuration_fails_closed(self):
|
||
|
|
for missing in ["WOLIN_AUTH_PASSWORD_HASH", "WOLIN_SESSION_SECRET"]:
|
||
|
|
with self.assertRaises(RuntimeError):
|
||
|
|
make_app(removed=(missing,))
|
||
|
|
for overrides in [{"WOLIN_SESSION_SECRET": "short"}, {"WOLIN_AUTH_PASSWORD_HASH": "plaintext"},
|
||
|
|
{"WOLIN_AUTH_PASSWORD_HASH": PASSWORD_HASH.replace("16384", "32768")},
|
||
|
|
{"AUTH_COOKIE_SECURE": "maybe"}, {"AUTH_ENABLED": "maybe"}, {"WOLIN_AUTH_USER": ""}]:
|
||
|
|
with self.assertRaises(RuntimeError):
|
||
|
|
make_app(overrides)
|
||
|
|
|
||
|
|
def test_rate_limiter_memory_is_bounded(self):
|
||
|
|
limiter = web_auth._LoginLimiter()
|
||
|
|
for index in range(4100):
|
||
|
|
limiter.reserve(str(index))
|
||
|
|
self.assertLessEqual(len(limiter.entries), 4096)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|