Compare commits
2 Commits
60f98756c4
...
7fe2dee03c
| Author | SHA1 | Date | |
|---|---|---|---|
| 7fe2dee03c | |||
| 6fef7c1dc7 |
+8
-4
@@ -82,10 +82,14 @@ def get_config() -> Config:
|
|||||||
database_url = os.environ.get("DATABASE_URL", "")
|
database_url = os.environ.get("DATABASE_URL", "")
|
||||||
if role != "home" and not database_url:
|
if role != "home" and not database_url:
|
||||||
raise SystemExit("DATABASE_URL is required unless APP_ROLE=home")
|
raise SystemExit("DATABASE_URL is required unless APP_ROLE=home")
|
||||||
# Where the pages this deployment does not serve actually live. A "full" deployment serves
|
# Where the pages this deployment does not serve actually live -- the "full" deployment that
|
||||||
# them itself, so its own base URL is the right answer. The OIDC issuer defaults to the same
|
# owns registration, profile and admin. A "full" deployment serves them itself, so its own
|
||||||
# host, so moving the auth service means changing AUTH_BASE_URL alone.
|
# base URL is the right answer. This is NOT the identity provider: OIDC_ISSUER points at
|
||||||
|
# Authentik, which lives on a different host.
|
||||||
auth_base_url = os.environ.get("AUTH_BASE_URL", base_url).rstrip("/")
|
auth_base_url = os.environ.get("AUTH_BASE_URL", base_url).rstrip("/")
|
||||||
|
oidc_issuer = os.environ.get("OIDC_ISSUER", "").rstrip("/")
|
||||||
|
if role != "home" and not oidc_issuer:
|
||||||
|
raise SystemExit("OIDC_ISSUER is required unless APP_ROLE=home")
|
||||||
return Config(
|
return Config(
|
||||||
role=role,
|
role=role,
|
||||||
auth_base_url=auth_base_url,
|
auth_base_url=auth_base_url,
|
||||||
@@ -99,7 +103,7 @@ def get_config() -> Config:
|
|||||||
default_ipa_group=os.environ.get("DEFAULT_IPA_GROUP", "sticknife_users"),
|
default_ipa_group=os.environ.get("DEFAULT_IPA_GROUP", "sticknife_users"),
|
||||||
app_admin_emails=_csv("APP_ADMIN_EMAILS"),
|
app_admin_emails=_csv("APP_ADMIN_EMAILS"),
|
||||||
admin_groups=_csv("APP_ADMIN_GROUPS") or {"sticknife_admins"},
|
admin_groups=_csv("APP_ADMIN_GROUPS") or {"sticknife_admins"},
|
||||||
oidc_issuer=os.environ.get("OIDC_ISSUER", f"{auth_base_url}/application/o/charon/").rstrip("/"),
|
oidc_issuer=oidc_issuer,
|
||||||
oidc_client_id=os.environ.get("OIDC_CLIENT_ID", ""),
|
oidc_client_id=os.environ.get("OIDC_CLIENT_ID", ""),
|
||||||
oidc_client_secret=os.environ.get("OIDC_CLIENT_SECRET", ""),
|
oidc_client_secret=os.environ.get("OIDC_CLIENT_SECRET", ""),
|
||||||
oidc_scopes=os.environ.get("OIDC_SCOPES", "openid profile email groups"),
|
oidc_scopes=os.environ.get("OIDC_SCOPES", "openid profile email groups"),
|
||||||
|
|||||||
+2
-1
@@ -92,6 +92,7 @@ def groups_from_claims(claims: dict[str, Any]) -> set[str]:
|
|||||||
|
|
||||||
def logout_url(config: Config) -> str:
|
def logout_url(config: Config) -> str:
|
||||||
provider = discover(config)
|
provider = discover(config)
|
||||||
endpoint = provider.end_session_endpoint or f"{config.auth_base_url}/if/session-end/"
|
# Authentik advertises {issuer}/end-session/; fall back to that shape if discovery omits it.
|
||||||
|
endpoint = provider.end_session_endpoint or f"{config.oidc_issuer}/end-session/"
|
||||||
query = urllib.parse.urlencode({"post_logout_redirect_uri": config.base_url})
|
query = urllib.parse.urlencode({"post_logout_redirect_uri": config.base_url})
|
||||||
return f"{endpoint}?{query}"
|
return f"{endpoint}?{query}"
|
||||||
|
|||||||
@@ -18,5 +18,5 @@ Host of the library and related services.
|
|||||||
|
|
||||||
# the paragraph in the detail panel
|
# the paragraph in the detail panel
|
||||||
long = """
|
long = """
|
||||||
Xenia is the goddess of hospitality, and the librarian of the pantheon. She runs sticknife library, a home for books, journals, and comics.
|
Xenia is the goddess of hospitality, and hosts the home page for Sticknife. She also runs sticknife library, a home for books, journals, and comics.
|
||||||
"""
|
"""
|
||||||
|
|||||||
+68
-13
@@ -1,7 +1,9 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import base64
|
||||||
import hashlib
|
import hashlib
|
||||||
import html
|
import html
|
||||||
|
import json
|
||||||
import mimetypes
|
import mimetypes
|
||||||
import urllib.parse
|
import urllib.parse
|
||||||
from http import HTTPStatus
|
from http import HTTPStatus
|
||||||
@@ -26,7 +28,16 @@ STATIC = ROOT / "static"
|
|||||||
|
|
||||||
#: The only routes a role="home" deployment answers. Everything else belongs to the deployment
|
#: The only routes a role="home" deployment answers. Everything else belongs to the deployment
|
||||||
#: that holds the database and the OIDC client, and is reached through config.auth_base_url.
|
#: that holds the database and the OIDC client, and is reached through config.auth_base_url.
|
||||||
HOME_ROUTES = {("GET", "/"), ("GET", "/trusted-domains")}
|
#: Served by a home deployment itself. Everything else is forwarded to the deployment that owns
|
||||||
|
#: identity. The auth routes are here so a home site can hold its own Authentik session: it needs
|
||||||
|
#: to know who you are to show "My Account" and "Admin", even though it owns no user table.
|
||||||
|
HOME_ROUTES = {
|
||||||
|
("GET", "/"),
|
||||||
|
("GET", "/trusted-domains"),
|
||||||
|
("GET", "/auth/start"),
|
||||||
|
("GET", "/auth/callback"),
|
||||||
|
("POST", "/logout"),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
def esc(value: object) -> str:
|
def esc(value: object) -> str:
|
||||||
@@ -38,6 +49,11 @@ class WebApp:
|
|||||||
self.config = config
|
self.config = config
|
||||||
# A home deployment has no database credentials at all, so there is nothing to point at.
|
# A home deployment has no database credentials at all, so there is nothing to point at.
|
||||||
self.db = None if config.role == "home" else Database(config.database_url)
|
self.db = None if config.role == "home" else Database(config.database_url)
|
||||||
|
#: Whether this deployment can run the OIDC flow itself. A home site without these falls
|
||||||
|
#: back to handing sign-in off to auth_base_url.
|
||||||
|
self.oidc_ready = bool(
|
||||||
|
config.oidc_issuer and config.oidc_client_id and config.oidc_client_secret
|
||||||
|
)
|
||||||
|
|
||||||
def dispatch(self, handler) -> None:
|
def dispatch(self, handler) -> None:
|
||||||
parsed = urlparse(handler.path)
|
parsed = urlparse(handler.path)
|
||||||
@@ -87,16 +103,21 @@ class WebApp:
|
|||||||
return {key: values[0] for key, values in parse_qs(raw).items()}
|
return {key: values[0] for key, values in parse_qs(raw).items()}
|
||||||
|
|
||||||
def current_user(self, handler) -> dict[str, str | None] | None:
|
def current_user(self, handler) -> dict[str, str | None] | None:
|
||||||
if self.db is None:
|
|
||||||
return None
|
|
||||||
cookie = SimpleCookie(handler.headers.get("Cookie"))
|
cookie = SimpleCookie(handler.headers.get("Cookie"))
|
||||||
morsel = cookie.get("snreg_session")
|
morsel = cookie.get("snreg_session")
|
||||||
if not morsel:
|
if not morsel:
|
||||||
return None
|
return None
|
||||||
user_id = unsign(morsel.value, self.config.secret)
|
value = unsign(morsel.value, self.config.secret)
|
||||||
if not user_id:
|
if not value:
|
||||||
return None
|
return None
|
||||||
return self.db.one(f"SELECT * FROM snreg_users WHERE id = {sql_literal(user_id)}")
|
if self.db is None:
|
||||||
|
# No user table here, so the claims we care about travel in the cookie itself. It is
|
||||||
|
# signed, so the contents are tamper-evident; nothing secret goes in.
|
||||||
|
try:
|
||||||
|
return json.loads(base64.urlsafe_b64decode(value.encode()).decode())
|
||||||
|
except Exception:
|
||||||
|
return None
|
||||||
|
return self.db.one(f"SELECT * FROM snreg_users WHERE id = {sql_literal(value)}")
|
||||||
|
|
||||||
def require_user(self, handler) -> dict[str, str | None] | None:
|
def require_user(self, handler) -> dict[str, str | None] | None:
|
||||||
user = self.current_user(handler)
|
user = self.current_user(handler)
|
||||||
@@ -168,9 +189,22 @@ class WebApp:
|
|||||||
|
|
||||||
def nav(self, user: dict[str, str | None] | None) -> str:
|
def nav(self, user: dict[str, str | None] | None) -> str:
|
||||||
if not user:
|
if not user:
|
||||||
return f'<a class="nav-button" href="{self.off_box("/auth/start")}?next=/profile">Sign in</a>'
|
# Sign in here when this deployment can run the flow itself; otherwise hand off to the
|
||||||
admin = '<a class="nav-button" href="/admin">Admin</a>' if user.get("is_admin") == "t" else ""
|
# deployment that owns identity.
|
||||||
return f'<a class="nav-button" href="/">Home</a><a class="nav-button" href="/profile">Profile</a>{admin}<form method="post" action="/logout"><button class="nav-button">Sign out</button></form>'
|
if self.oidc_ready:
|
||||||
|
start, nxt = "/auth/start", "/" if self.config.role == "home" else "/profile"
|
||||||
|
else:
|
||||||
|
start, nxt = self.off_box("/auth/start"), "/profile"
|
||||||
|
return f'<a class="nav-button" href="{start}?next={nxt}">Sign in</a>'
|
||||||
|
# off_box leaves these relative on a full deployment and absolute on a home one, so the
|
||||||
|
# same markup serves both.
|
||||||
|
admin = (
|
||||||
|
f'<a class="nav-button" href="{self.off_box("/admin")}">Admin</a>'
|
||||||
|
if user.get("is_admin") == "t"
|
||||||
|
else ""
|
||||||
|
)
|
||||||
|
account = f'<a class="nav-button" href="{self.off_box("/profile")}">My Account</a>'
|
||||||
|
return f'<a class="nav-button" href="/">Home</a>{account}{admin}<form method="post" action="/logout"><button class="nav-button">Sign out</button></form>'
|
||||||
|
|
||||||
def redirect(self, handler, location: str) -> None:
|
def redirect(self, handler, location: str) -> None:
|
||||||
handler.send_response(303)
|
handler.send_response(303)
|
||||||
@@ -183,6 +217,22 @@ class WebApp:
|
|||||||
cookie += "; Secure"
|
cookie += "; Secure"
|
||||||
handler.send_header("Set-Cookie", cookie)
|
handler.send_header("Set-Cookie", cookie)
|
||||||
|
|
||||||
|
def set_home_session(self, handler, claims: dict) -> None:
|
||||||
|
"""Session for a deployment with no user table, built straight from the OIDC claims."""
|
||||||
|
groups = groups_from_claims(claims)
|
||||||
|
email = str(claims.get("email") or "").lower()
|
||||||
|
payload = {
|
||||||
|
"username": str(claims.get("preferred_username") or email.split("@", 1)[0] or ""),
|
||||||
|
"full_name": str(claims.get("name") or ""),
|
||||||
|
"email": email,
|
||||||
|
# Same rule the full deployment applies in upsert_oidc_user.
|
||||||
|
"is_admin": "t"
|
||||||
|
if email in self.config.app_admin_emails or groups & self.config.admin_groups
|
||||||
|
else "f",
|
||||||
|
}
|
||||||
|
encoded = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode()
|
||||||
|
self.set_session(handler, encoded)
|
||||||
|
|
||||||
def off_box(self, path: str) -> str:
|
def off_box(self, path: str) -> str:
|
||||||
"""Absolute URL for a page this deployment does not serve; unchanged when it does.
|
"""Absolute URL for a page this deployment does not serve; unchanged when it does.
|
||||||
|
|
||||||
@@ -239,9 +289,11 @@ class WebApp:
|
|||||||
if not self.config.oidc_client_id or not self.config.oidc_client_secret:
|
if not self.config.oidc_client_id or not self.config.oidc_client_secret:
|
||||||
return self.render(handler, "Sign in unavailable", "<p>Authentik OIDC is not configured for Charon yet.</p>", HTTPStatus.SERVICE_UNAVAILABLE)
|
return self.render(handler, "Sign in unavailable", "<p>Authentik OIDC is not configured for Charon yet.</p>", HTTPStatus.SERVICE_UNAVAILABLE)
|
||||||
query = parse_qs(urlparse(handler.path).query)
|
query = parse_qs(urlparse(handler.path).query)
|
||||||
next_path = query.get("next", ["/profile"])[0]
|
# A home deployment does not serve /profile, so land back on its own front page.
|
||||||
|
default_next = "/" if self.config.role == "home" else "/profile"
|
||||||
|
next_path = query.get("next", [default_next])[0]
|
||||||
if not next_path.startswith("/") or next_path.startswith("//"):
|
if not next_path.startswith("/") or next_path.startswith("//"):
|
||||||
next_path = "/profile"
|
next_path = default_next
|
||||||
state = sign(next_path, self.config.secret)
|
state = sign(next_path, self.config.secret)
|
||||||
self.redirect(handler, authorization_url(self.config, state, next_path))
|
self.redirect(handler, authorization_url(self.config, state, next_path))
|
||||||
|
|
||||||
@@ -255,12 +307,15 @@ class WebApp:
|
|||||||
try:
|
try:
|
||||||
token = exchange_code(self.config, code)
|
token = exchange_code(self.config, code)
|
||||||
claims = userinfo(self.config, token["access_token"])
|
claims = userinfo(self.config, token["access_token"])
|
||||||
user = self.upsert_oidc_user(claims)
|
user = None if self.db is None else self.upsert_oidc_user(claims)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
return self.render(handler, "Sign in failed", f"<p>{esc(exc)}</p>", HTTPStatus.BAD_GATEWAY)
|
return self.render(handler, "Sign in failed", f"<p>{esc(exc)}</p>", HTTPStatus.BAD_GATEWAY)
|
||||||
handler.send_response(303)
|
handler.send_response(303)
|
||||||
handler.send_header("Location", next_path)
|
handler.send_header("Location", next_path)
|
||||||
self.set_session(handler, user["id"] or "")
|
if user is None:
|
||||||
|
self.set_home_session(handler, claims)
|
||||||
|
else:
|
||||||
|
self.set_session(handler, user["id"] or "")
|
||||||
handler.end_headers()
|
handler.end_headers()
|
||||||
|
|
||||||
def upsert_oidc_user(self, claims: dict) -> dict[str, str | None]:
|
def upsert_oidc_user(self, claims: dict) -> dict[str, str | None]:
|
||||||
|
|||||||
Reference in New Issue
Block a user