128 lines
3.6 KiB
Python
128 lines
3.6 KiB
Python
"""Central deny-by-default policy for sensitive HTTP surfaces."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
|
|
POLICY = "ewoooc_sensitive_http_access_policy_v1"
|
|
|
|
# Entire blueprints that expose business data or operator actions.
|
|
SESSION_REQUIRED_BLUEPRINTS = frozenset({
|
|
"auto_import",
|
|
"category",
|
|
"cicd",
|
|
"crawler",
|
|
"crawler_management",
|
|
"vendor",
|
|
})
|
|
|
|
# Mixed public/private blueprints use endpoint-level protection.
|
|
SESSION_REQUIRED_ENDPOINTS = frozenset({
|
|
"alert.alert_status",
|
|
"alert.test_alert",
|
|
"code_review.api_history",
|
|
"code_review.api_status",
|
|
"code_review.dashboard",
|
|
"misc.test_url",
|
|
"openclaw_bot.set_webhook",
|
|
"openclaw_bot.webhook_info",
|
|
"sales.abc_analysis_detail",
|
|
"system_public.get_logs_api",
|
|
"system_public.settings",
|
|
"system_public.show_logs",
|
|
"system_public.system_settings_page",
|
|
})
|
|
|
|
# Every anonymous GET must be intentional and small enough for public exposure.
|
|
INTENTIONALLY_PUBLIC_ENDPOINTS = frozenset({
|
|
"bot_api.bot_api_status",
|
|
"elephant_alpha.status",
|
|
"misc.brand_assets",
|
|
"system_public.favicon",
|
|
"system_public.health_check",
|
|
"system_public.webcrumbs_asset_proxy",
|
|
"user_bp.get_password_reqs",
|
|
})
|
|
|
|
# Metrics stay anonymous only on explicitly trusted transport networks. The edge
|
|
# proxy applies the same allowlist, while this policy protects direct app access.
|
|
INTERNAL_ONLY_ENDPOINTS = frozenset({
|
|
"system_public.prometheus_metrics",
|
|
})
|
|
INTERNAL_METRICS_NETWORKS = tuple(
|
|
ipaddress.ip_network(cidr)
|
|
for cidr in (
|
|
"127.0.0.0/8",
|
|
"172.16.0.0/12",
|
|
"192.168.0.110/32",
|
|
"::1/128",
|
|
)
|
|
)
|
|
|
|
|
|
def is_session_required(endpoint: str | None, blueprint: str | None) -> bool:
|
|
return bool(
|
|
blueprint in SESSION_REQUIRED_BLUEPRINTS
|
|
or endpoint in SESSION_REQUIRED_ENDPOINTS
|
|
)
|
|
|
|
|
|
def is_internal_metrics_request(
|
|
remote_addr: str | None,
|
|
x_real_ip: str | None = None,
|
|
x_forwarded_for: str | None = None,
|
|
) -> bool:
|
|
"""Return whether the effective client belongs to the metrics allowlist."""
|
|
try:
|
|
peer_address = ipaddress.ip_address((remote_addr or "").strip())
|
|
except ValueError:
|
|
return False
|
|
if not any(peer_address in network for network in INTERNAL_METRICS_NETWORKS):
|
|
return False
|
|
|
|
forwarded_ip = (x_forwarded_for or "").split(",", 1)[0].strip()
|
|
candidate = (
|
|
(x_real_ip or "").strip()
|
|
or forwarded_ip
|
|
or str(peer_address)
|
|
)
|
|
try:
|
|
address = ipaddress.ip_address(candidate)
|
|
except ValueError:
|
|
return False
|
|
return any(address in network for network in INTERNAL_METRICS_NETWORKS)
|
|
|
|
|
|
def enforce_http_access_policy():
|
|
"""Protect sensitive routes before their route function executes."""
|
|
from flask import jsonify, request
|
|
|
|
if request.endpoint in INTERNAL_ONLY_ENDPOINTS:
|
|
allowed = is_internal_metrics_request(
|
|
request.remote_addr,
|
|
request.headers.get("X-Real-IP"),
|
|
request.headers.get("X-Forwarded-For"),
|
|
)
|
|
if not allowed:
|
|
return jsonify({"error": "not_found"}), 404
|
|
return None
|
|
|
|
if not is_session_required(request.endpoint, request.blueprint):
|
|
return None
|
|
from auth import require_authenticated_request
|
|
|
|
return require_authenticated_request()
|
|
|
|
|
|
__all__ = [
|
|
"POLICY",
|
|
"SESSION_REQUIRED_BLUEPRINTS",
|
|
"SESSION_REQUIRED_ENDPOINTS",
|
|
"INTENTIONALLY_PUBLIC_ENDPOINTS",
|
|
"INTERNAL_ONLY_ENDPOINTS",
|
|
"INTERNAL_METRICS_NETWORKS",
|
|
"enforce_http_access_policy",
|
|
"is_internal_metrics_request",
|
|
"is_session_required",
|
|
]
|