Implement metadata for RestrictedForward, set by /auth/api/forward endpoint when returning the app. Use this to implement support for time-based reauth requirement.
This commit is contained in:
@@ -0,0 +1,47 @@
|
||||
"""Utility functions for HTML manipulation."""
|
||||
|
||||
import re
|
||||
|
||||
|
||||
def patch_html_data_attrs(html: bytes, **data_attrs: str) -> bytes:
|
||||
"""Patch HTML by adding data attributes to the <html> tag.
|
||||
|
||||
If an <html> tag exists, adds data attributes to it.
|
||||
If no <html> tag exists, prepends one with the data attributes.
|
||||
|
||||
Args:
|
||||
html: The HTML content as bytes
|
||||
**data_attrs: Key-value pairs for data attributes (e.g., mode='reauth')
|
||||
|
||||
Returns:
|
||||
Modified HTML as bytes
|
||||
|
||||
Examples:
|
||||
>>> patch_html_data_attrs(b'<html><body>test</body></html>', mode='reauth')
|
||||
b'<html data-mode="reauth"><body>test</body></html>'
|
||||
|
||||
>>> patch_html_data_attrs(b'<body>test</body>', mode='reauth')
|
||||
b'<html data-mode="reauth"><body>test</body>'
|
||||
"""
|
||||
if not data_attrs:
|
||||
return html
|
||||
|
||||
html_str = html.decode("utf-8")
|
||||
|
||||
# Build the data attributes string
|
||||
attrs_str = " ".join(f'data-{key}="{value}"' for key, value in data_attrs.items())
|
||||
|
||||
# Check if there's an <html> tag (case-insensitive, may have existing attributes)
|
||||
html_tag_pattern = re.compile(r"<html([^>]*)>", re.IGNORECASE)
|
||||
match = html_tag_pattern.search(html_str)
|
||||
|
||||
if match:
|
||||
# Insert data attributes into existing <html> tag
|
||||
existing_attrs = match.group(1)
|
||||
new_tag = f"<html{existing_attrs} {attrs_str}>"
|
||||
html_str = html_tag_pattern.sub(new_tag, html_str, count=1)
|
||||
else:
|
||||
# Prepend <html> tag with data attributes
|
||||
html_str = f"<html {attrs_str}>" + html_str
|
||||
|
||||
return html_str.encode("utf-8")
|
||||
@@ -0,0 +1,27 @@
|
||||
"""Utility functions for session validation and checking."""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from ..db import Session
|
||||
from .timeutil import parse_duration
|
||||
|
||||
|
||||
def check_session_age(session: Session, max_age: str | None) -> bool:
|
||||
"""Check if a session satisfies the max_age requirement.
|
||||
|
||||
Args:
|
||||
session: The session record to check
|
||||
max_age: Maximum age string (e.g., "5m", "1h", "30s") or None
|
||||
|
||||
Returns:
|
||||
True if session is recent enough or max_age is None, False if too old
|
||||
|
||||
Raises:
|
||||
ValueError: If max_age format is invalid
|
||||
"""
|
||||
if not max_age:
|
||||
return True
|
||||
|
||||
max_age_delta = parse_duration(max_age)
|
||||
time_since_auth = datetime.now(timezone.utc) - session.renewed
|
||||
return time_since_auth <= max_age_delta
|
||||
@@ -0,0 +1,47 @@
|
||||
"""Utility functions for parsing time durations."""
|
||||
|
||||
import re
|
||||
from datetime import timedelta
|
||||
|
||||
|
||||
def parse_duration(duration_str: str) -> timedelta:
|
||||
"""Parse a duration string into a timedelta.
|
||||
|
||||
Supports units: s, m, min, h, d
|
||||
Examples: "30s", "5m", "5min", "2h", "1d"
|
||||
|
||||
Args:
|
||||
duration_str: A string like "30s", "5m", "2h"
|
||||
|
||||
Returns:
|
||||
A timedelta object
|
||||
|
||||
Raises:
|
||||
ValueError: If the format is invalid
|
||||
"""
|
||||
duration_str = duration_str.strip().lower()
|
||||
|
||||
# Pattern matches: number + unit
|
||||
# Units: s (seconds), m/min (minutes), h (hours), d (days)
|
||||
pattern = r"^(\d+(?:\.\d+)?)(s|m|min|h|d)$"
|
||||
match = re.match(pattern, duration_str)
|
||||
|
||||
if not match:
|
||||
raise ValueError(
|
||||
f"Invalid duration format: '{duration_str}'. "
|
||||
"Expected format like '30s', '5m', '5min', '2h', or '1d'"
|
||||
)
|
||||
|
||||
value = float(match.group(1))
|
||||
unit = match.group(2)
|
||||
|
||||
if unit == "s":
|
||||
return timedelta(seconds=value)
|
||||
elif unit in ("m", "min"):
|
||||
return timedelta(minutes=value)
|
||||
elif unit == "h":
|
||||
return timedelta(hours=value)
|
||||
elif unit == "d":
|
||||
return timedelta(days=value)
|
||||
else:
|
||||
raise ValueError(f"Unsupported time unit: {unit}")
|
||||
Reference in New Issue
Block a user