Files
snek/src/snek/system/middleware.py
T
2025-06-06 03:41:39 +02:00

76 lines
2.5 KiB
Python

# Written by retoor@molodetz.nl
# This code provides middleware functions for an aiohttp server to manage and modify CORS (Cross-Origin Resource Sharing) headers.
# Imports from 'aiohttp' library are used to create middleware; they are not part of Python's standard library.
# MIT License: This code is distributed under the MIT License.
from aiohttp import web
import secrets
csp_policy = (
"default-src 'self'; "
"script-src 'self' https://*.cloudflare.com https://molodetz.nl 'nonce-{nonce}'; "
"style-src 'self' https://*.cloudflare.com https://molodetz.nl; "
"img-src 'self' https://*.cloudflare.com https://molodetz.nl data:; "
"connect-src 'self' https://*.cloudflare.com https://molodetz.nl;"
)
def generate_nonce():
return secrets.token_hex(16)
@web.middleware
async def csp_middleware(request, handler):
response = await handler(request)
return response
nonce = generate_nonce()
response.headers['Content-Security-Policy'] = csp_policy.format(nonce=nonce)
return response
@web.middleware
async def no_cors_middleware(request, handler):
response = await handler(request)
response.headers.pop("Access-Control-Allow-Origin", None)
return response
@web.middleware
async def cors_allow_middleware(request, handler):
response = await handler(request)
response.headers["Access-Control-Allow-Origin"] = "*"
response.headers["Access-Control-Allow-Methods"] = (
"GET, POST, OPTIONS, PUT, DELETE, MOVE, COPY, HEAD, LOCK, UNLOCK, PATCH, PROPFIND"
)
response.headers["Access-Control-Allow-Headers"] = "*"
response.headers["Access-Control-Allow-Credentials"] = "true"
return response
@web.middleware
async def auth_middleware(request, handler):
request["user"] = None
if request.session.get("uid") and request.session.get("logged_in"):
request["user"] = await request.app.services.user.get(
uid=request.app.session.get("uid")
)
return await handler(request)
@web.middleware
async def cors_middleware(request, handler):
if request.headers.get("Allow"):
return await handler(request)
response = await handler(request)
if request.headers.get("Allow"):
return response
response.headers["Access-Control-Allow-Origin"] = "*"
response.headers["Access-Control-Allow-Methods"] = "GET, POST, PUT, DELETE, OPTIONS"
response.headers["Access-Control-Allow-Headers"] = "*"
response.headers["Access-Control-Allow-Credentials"] = "true"
return response