hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 33 Packages Projects Releases Wiki Activity

fix(dashboard-auth): pin OIDC discovery to the configured issuer origin

Browse Source 
_fetch_discovery followed redirects but only pinned the document's
self-asserted issuer field, so one cleartext or attacker-hosted hop
could serve a forged document claiming the configured issuer with
attacker jwks_uri and token_endpoint. Verify then accepted
attacker-signed ID tokens and the code exchange POSTed the client
secret to the attacker's token endpoint.

The resolved response.url must now share the configured issuer's
origin (scheme, host, port with default-port normalisation) before
the body is parsed. Same-origin canonicalisation redirects still
pass, and the issuer-field pin remains as the misconfig check it is.
This commit is contained in:
beardthelion
2026-09-19 15:06:20 -05:00
committed by Teknium
parent ba3b1eb3c3
commit 07c1953ea1
2 changed files with 173 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
plugins/dashboard_auth/self_hosted/__init__.py
View File

@@ -64,6 +64,15 @@ def _require_https_or_loopback(url: str, *, field: str) -> str:
raise ProviderError(f"OIDC {field} must be https:// (or http on localhost), got {url!r}")
def _origin(url: str) -> tuple:
"""``(scheme, hostname, port)`` for an origin compare, with default ports
normalised so ``https://h`` and ``https://h:443`` are the same origin."""
parts = urllib.parse.urlparse(url)
scheme = parts.scheme.lower()
return (scheme, (parts.hostname or "").lower(),
parts.port or {"https": 443, "http": 80}.get(scheme))
class SelfHostedOIDCProvider(JwtOAuthProvider):
"""Generic self-hosted OpenID Connect provider (authorization-code + PKCE)."""
@@ -188,14 +197,23 @@ class SelfHostedOIDCProvider(JwtOAuthProvider):
try:
# follow_redirects=True: many IDPs answer discovery with a 3xx (Authentik
# canonicalises .well-known; proxies upgrade http→https) and httpx defaults to
# not following. Safe because the issuer pin and HTTPS checks below validate the
# *resolved* document. The token/revocation POSTs deliberately do NOT follow
# not following. The token/revocation POSTs deliberately do NOT follow
# redirects (they carry an auth code / refresh token).
response = httpx.get(url, headers=JSON_HEADERS, timeout=_DISCOVERY_TIMEOUT_SEC, follow_redirects=True)
except httpx.RequestError as exc:
raise ProviderError(f"OIDC discovery unreachable: {exc}") from exc
if response.status_code != 200:
raise ProviderError(f"OIDC discovery returned {response.status_code} for {url!r}")
# Resolved-origin pin: the document only counts as the IDP's when the url that
# actually served it shares the configured issuer's origin. The ``issuer`` field
# inside the body is attacker-controlled content and cannot prove where the
# document came from — a single cleartext or attacker-hosted redirect hop could
# otherwise serve a forged document asserting the configured issuer with
# attacker jwks_uri / token_endpoint.
if _origin(str(response.url)) != _origin(self._issuer):
raise ProviderError(
f"OIDC discovery resolved to {response.url}, outside the configured "
f"issuer's origin ({self._issuer!r})")
payload = parse_json_body(response)
if not payload:
raise ProviderError("OIDC discovery returned a non-JSON body")

154
tests/plugins/dashboard_auth/test_self_hosted_provider.py
View File

@@ -220,9 +220,10 @@ class TestDiscovery:
issuer=_ISSUER, client_id=_CLIENT_ID
)
def _mock_get(self, status_code, body, *, ctype="application/json"):
def _mock_get(self, status_code, body, *, ctype="application/json", url=None):
resp = MagicMock(spec=httpx.Response)
resp.status_code = status_code
resp.url = httpx.URL(url or f"{_ISSUER}/.well-known/openid-configuration")
resp.json = MagicMock(return_value=body)
resp.text = json.dumps(body) if isinstance(body, dict) else str(body)
resp.headers = {"content-type": ctype}
@@ -245,6 +246,60 @@ class TestDiscovery:
assert mock_get.call_count == 1
assert disco2 is disco1
def test_redirect_landing_off_origin_rejected(self):
"""The resolved url is the trust anchor, not the body's self-asserted
issuer: a redirect to an attacker origin serving a document that claims
the configured issuer (with attacker jwks_uri/token_endpoint) must fail."""
p = self._provider()
forged = {
**_DISCOVERY_DOC,
"jwks_uri": "https://attacker.example/jwks",
"token_endpoint": "https://attacker.example/token",
}
resp = self._mock_get(
200, forged, url="https://attacker.example/openid-configuration"
)
with patch(
"plugins.dashboard_auth.self_hosted.httpx.get", return_value=resp
):
with pytest.raises(ProviderError, match="origin"):
p._fetch_discovery()
def test_redirect_landing_on_cleartext_rejected(self):
p = self._provider()
resp = self._mock_get(
200, dict(_DISCOVERY_DOC), url="http://auth.example.com/discovery"
)
with patch(
"plugins.dashboard_auth.self_hosted.httpx.get", return_value=resp
):
with pytest.raises(ProviderError, match="origin"):
p._fetch_discovery()
def test_same_origin_redirect_allowed(self):
"""Canonicalisation redirects on the issuer's own origin still pass."""
p = self._provider()
resp = self._mock_get(
200, dict(_DISCOVERY_DOC),
url="https://auth.example.com/.well-known/openid-configuration/application/o/hermes",
)
with patch(
"plugins.dashboard_auth.self_hosted.httpx.get", return_value=resp
):
disco = p._fetch_discovery()
assert disco["token_endpoint"] == f"{_ISSUER}/token"
def test_explicit_default_port_is_same_origin(self):
"""https://host:443 must compare equal to https://host."""
p = self._provider()
resp = self._mock_get(
200, dict(_DISCOVERY_DOC), url="https://auth.example.com:443/x"
)
with patch(
"plugins.dashboard_auth.self_hosted.httpx.get", return_value=resp
):
assert p._fetch_discovery()["issuer"] == _ISSUER
# ---------------------------------------------------------------------------
# OIDC discovery against a REAL HTTP server that redirects (regression)
@@ -280,6 +335,103 @@ class TestDiscoveryRealRedirect:
thread.start()
return httpd, port
def _handler(self, routes):
"""Build a request handler serving {path: (status, headers, body_bytes)}."""
import http.server
class _H(http.server.BaseHTTPRequestHandler):
def do_GET(self):
status, headers, body = self.routes.get(self.path, (404, {}, b""))
self.send_response(status)
for k, v in headers.items():
self.send_header(k, v)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass
_H.routes = routes
return _H
def test_real_same_origin_redirect_succeeds(self):
httpd, port = self._serve(self._handler({}))
try:
issuer = f"http://127.0.0.1:{port}"
doc = json.dumps({
"issuer": issuer,
"authorization_endpoint": f"{issuer}/authorize",
"token_endpoint": f"{issuer}/token",
"jwks_uri": f"{issuer}/jwks",
}).encode()
httpd.RequestHandlerClass.routes = {
"/.well-known/openid-configuration": (
302, {"Location": f"{issuer}/canonical"}, b""),
"/canonical": (200, {"Content-Type": "application/json"}, doc),
}
p = oidc_plugin.SelfHostedOIDCProvider(issuer=issuer, client_id=_CLIENT_ID)
disco = p._fetch_discovery()
assert disco["issuer"] == issuer
finally:
httpd.shutdown()
def test_real_redirect_to_other_origin_rejected(self):
"""A 302 to a different origin (here: another loopback port) serving a
forged document that claims the issuer must fail before parsing."""
forge_httpd, forge_port = self._serve(self._handler({}))
redirect_httpd, redirect_port = self._serve(self._handler({}))
try:
issuer = f"http://127.0.0.1:{redirect_port}"
forged = json.dumps({
"issuer": issuer, # self-asserted, attacker-controlled
"authorization_endpoint": "https://attacker.example/authorize",
"token_endpoint": "https://attacker.example/token",
"jwks_uri": "https://attacker.example/jwks",
}).encode()
forge_httpd.RequestHandlerClass.routes = {
"/doc": (200, {"Content-Type": "application/json"}, forged),
}
redirect_httpd.RequestHandlerClass.routes = {
"/.well-known/openid-configuration": (
302, {"Location": f"http://127.0.0.1:{forge_port}/doc"}, b""),
}
p = oidc_plugin.SelfHostedOIDCProvider(issuer=issuer, client_id=_CLIENT_ID)
with pytest.raises(ProviderError, match="origin"):
p._fetch_discovery()
finally:
forge_httpd.shutdown()
redirect_httpd.shutdown()
def test_start_login_rejects_forged_discovery_through_real_redirect(self):
"""Consumer-level e2e: start_login goes through _get_discovery, so the
origin pin must stop the forged doc before any authorize URL is built,
and before exchange_token could POST the client_secret to the forged
token_endpoint."""
forge_httpd, forge_port = self._serve(self._handler({}))
redirect_httpd, redirect_port = self._serve(self._handler({}))
try:
issuer = f"http://127.0.0.1:{redirect_port}"
forged = json.dumps({
"issuer": issuer,
"authorization_endpoint": "https://attacker.example/authorize",
"token_endpoint": "https://attacker.example/token",
"jwks_uri": "https://attacker.example/jwks",
}).encode()
forge_httpd.RequestHandlerClass.routes = {
"/doc": (200, {"Content-Type": "application/json"}, forged),
}
redirect_httpd.RequestHandlerClass.routes = {
"/.well-known/openid-configuration": (
302, {"Location": f"http://127.0.0.1:{forge_port}/doc"}, b""),
}
p = oidc_plugin.SelfHostedOIDCProvider(issuer=issuer, client_id=_CLIENT_ID)
with pytest.raises(ProviderError, match="origin"):
p.start_login(redirect_uri="https://dash.example.com/auth/callback")
finally:
forge_httpd.shutdown()
redirect_httpd.shutdown()
# ---------------------------------------------------------------------------
# start_login