coryHawkvelt/gatehouse-api
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #37 from CoryHawkless/oidc-uplift

Browse Source 
OIDC uplift
This commit is contained in:
CoryHawkless
2026-05-19 14:48:58 +09:30
committed by GitHub
parent 32d517ea08 803bf4f4f2
commit f856aa5aea
31 changed files with 1808 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gatehouse_app/models/__init__.py
-2
View File
@@ -118,7 +118,6 @@ from gatehouse_app.models.zerotier import ( # noqa: F401
from gatehouse_app.models.superadmin import ( # noqa: F401
Superadmin,
SuperadminSession,
SuperadminSessionStatus,
)
from gatehouse_app.models.superadmin_audit_log import SuperadminAuditLog # noqa: F401
from gatehouse_app.models.security.user_security_policy import ( # noqa: F401
@@ -186,6 +185,5 @@ __all__ = [
# Superadmin
"Superadmin",
"SuperadminSession",
"SuperadminSessionStatus",
"SuperadminAuditLog",
]
gatehouse_app/models/oidc/oidc_client.py
+34
View File
@@ -1,4 +1,6 @@
"""OIDC Client model."""
from urllib.parse import urlparse
from gatehouse_app.extensions import db
from gatehouse_app.models.base import BaseModel
from gatehouse_app.utils.constants import OIDCGrantType, OIDCResponseType
@@ -21,6 +23,7 @@ class OIDCClient(BaseModel):
grant_types = db.Column(db.JSON, nullable=False) # Allowed grant types
response_types = db.Column(db.JSON, nullable=False) # Allowed response types
scopes = db.Column(db.JSON, nullable=False) # Allowed scopes
allowed_cors_origins = db.Column(db.JSON, nullable=True, default=None) # Per-client CORS origins
# Client metadata
logo_uri = db.Column(db.String(512), nullable=True)
@@ -81,6 +84,37 @@ class OIDCClient(BaseModel):
"""Check if a redirect URI is allowed for this client."""
return redirect_uri in self.redirect_uris
def get_effective_origins(self) -> list | None:
"""Get effective CORS origins for this client.
Returns None to signal "use global config", a derived list from
redirect_uris when "+" is present, or the configured list as-is.
"""
if self.allowed_cors_origins is None:
return None
if "+" in self.allowed_cors_origins:
origins = set()
for uri in self.redirect_uris:
parsed = urlparse(uri)
if parsed.scheme and parsed.hostname:
port = f":{parsed.port}" if parsed.port else ""
origins.add(f"{parsed.scheme}://{parsed.hostname}{port}")
return sorted(origins)
return list(self.allowed_cors_origins)
def is_origin_allowed(self, origin: str) -> bool | None:
"""Check if a browser origin is allowed for CORS.
Returns True/False when a per-client list is configured,
or None to defer to the global CORS policy.
"""
effective = self.get_effective_origins()
if effective is None:
return None
if "*" in effective:
return True
return origin in effective
def has_scope(self, scope: str) -> bool:
"""Check if client is allowed to request a specific scope."""
return scope in self.scopes
gatehouse_app/models/superadmin/__init__.py
+2 -2
View File
@@ -1,5 +1,5 @@
"""Superadmin models."""
from gatehouse_app.models.superadmin.superadmin import Superadmin
from gatehouse_app.models.superadmin.superadmin_session import SuperadminSession, SuperadminSessionStatus
from gatehouse_app.models.user.session import Session as SuperadminSession
__all__ = ["Superadmin", "SuperadminSession", "SuperadminSessionStatus"]
__all__ = ["Superadmin", "SuperadminSession"]
gatehouse_app/models/superadmin/superadmin.py
+8 -4
View File
@@ -23,11 +23,15 @@ class Superadmin(BaseModel):
is_active = db.Column(db.Boolean, default=True, nullable=False)
last_login_at = db.Column(db.DateTime, nullable=True)
# Relationship to sessions
# Relationship to sessions (unified model, scoped to superadmin owner_type)
sessions = db.relationship(
"SuperadminSession",
back_populates="superadmin",
cascade="all, delete-orphan"
"Session",
primaryjoin=(
"and_(Superadmin.id == foreign(Session.owner_id), "
"Session.owner_type == 'superadmin')"
),
cascade="all, delete-orphan",
lazy="dynamic",
)
# Relationship to audit logs
gatehouse_app/models/superadmin/superadmin_session.py
-80
View File
@@ -1,80 +0,0 @@
"""Superadmin session model."""
import logging
from datetime import datetime, timezone, timedelta
from gatehouse_app.extensions import db
from gatehouse_app.models.base import BaseModel
logger = logging.getLogger(__name__)
class SuperadminSessionStatus:
"""Session status constants."""
ACTIVE = "active"
REVOKED = "revoked"
EXPIRED = "expired"
class SuperadminSession(BaseModel):
"""Session model for superadmin authentication."""
__tablename__ = "superadmin_sessions"
superadmin_id = db.Column(
db.String(36),
db.ForeignKey("superadmins.id"),
nullable=False,
index=True
)
token = db.Column(db.String(255), unique=True, nullable=False, index=True)
expires_at = db.Column(db.DateTime, nullable=False)
last_activity_at = db.Column(
db.DateTime,
nullable=False,
default=lambda: datetime.now(timezone.utc)
)
ip_address = db.Column(db.String(45), nullable=True)
user_agent = db.Column(db.Text, nullable=True)
revoked_at = db.Column(db.DateTime, nullable=True)
revoked_reason = db.Column(db.String(255), nullable=True)
# Relationship
superadmin = db.relationship("Superadmin", back_populates="sessions")
def __repr__(self):
return f"<SuperadminSession superadmin_id={self.superadmin_id}>"
def is_active(self):
"""Check if session is currently active."""
now = datetime.now(timezone.utc)
expires_at = self.expires_at
if expires_at.tzinfo is None:
expires_at = expires_at.replace(tzinfo=timezone.utc)
return (
self.deleted_at is None
and self.revoked_at is None
and expires_at > now
)
def is_expired(self):
"""Check if session has expired."""
now = datetime.now(timezone.utc)
expires_at = self.expires_at
if expires_at.tzinfo is None:
expires_at = expires_at.replace(tzinfo=timezone.utc)
return now > expires_at
def revoke(self, reason: str = None):
"""Revoke the session."""
self.revoked_at = datetime.now(timezone.utc)
if reason:
self.revoked_reason = reason
from gatehouse_app import db
db.session.commit()
def to_dict(self, exclude=None):
"""Convert to dictionary, excluding sensitive fields."""
exclude = exclude or []
exclude.append("token")
return super().to_dict(exclude=exclude)
gatehouse_app/models/user/session.py
+54 -12
View File
@@ -3,15 +3,24 @@ from datetime import datetime, timedelta, timezone
from flask import current_app
from gatehouse_app.extensions import db
from gatehouse_app.models.base import BaseModel
from gatehouse_app.utils.constants import SessionStatus
from gatehouse_app.utils.constants import SessionStatus, SessionType
class Session(BaseModel):
"""Session model for tracking user sessions."""
"""Session model for tracking user and superadmin sessions."""
__tablename__ = "sessions"
user_id = db.Column(db.String(36), db.ForeignKey("users.id"), nullable=False, index=True)
# Owner discriminator — determines which table the owner_id references
owner_type = db.Column(
db.String(20), nullable=False, default=SessionType.USER, index=True
)
owner_id = db.Column(db.String(36), nullable=False, index=True)
# Legacy column kept for backward compatibility during migration;
# new code should use owner_id / owner_type.
user_id = db.Column(db.String(36), db.ForeignKey("users.id"), nullable=True, index=True)
token = db.Column(db.String(255), unique=True, nullable=False, index=True)
status = db.Column(db.Enum(SessionStatus), default=SessionStatus.ACTIVE, nullable=False)
@@ -34,21 +43,37 @@ class Session(BaseModel):
# Relationships
user = db.relationship("User", back_populates="sessions")
# Composite index for owner-scoped queries
__table_args__ = (
db.Index("ix_sessions_owner_type_owner_id", "owner_type", "owner_id"),
)
# ---- Convenience properties ------------------------------------------------
@property
def is_user(self):
return self.owner_type == SessionType.USER
@property
def is_superadmin(self):
return self.owner_type == SessionType.SUPERADMIN
# ---- Core methods ----------------------------------------------------------
def __repr__(self):
"""String representation of Session."""
return f"<Session user_id={self.user_id} status={self.status}>"
return f"<Session owner_type={self.owner_type} owner_id={self.owner_id} status={self.status}>"
def is_active(self):
"""Check if session is currently active.
Sessions are evaluated against two independent timeouts:
User sessions are evaluated against two independent timeouts:
- Idle timeout: expires if no request has been made within
SESSION_IDLE_TIMEOUT seconds (default 15 min).
- Absolute timeout: expires if SESSION_ABSOLUTE_TIMEOUT seconds
have elapsed since the session was created (default 8 h),
regardless of activity.
have elapsed since the session was created (default 8 h).
A session must satisfy *both* constraints to remain active.
Superadmin sessions use absolute timeout only (no idle timeout).
A session must satisfy *all* applicable constraints to remain active.
"""
now = datetime.now(timezone.utc)
created_at = self.created_at
@@ -59,12 +84,21 @@ class Session(BaseModel):
if last_activity_at.tzinfo is None:
last_activity_at = last_activity_at.replace(tzinfo=timezone.utc)
idle_timeout = current_app.config.get("SESSION_IDLE_TIMEOUT", 900)
absolute_timeout = current_app.config.get("SESSION_ABSOLUTE_TIMEOUT", 28800)
idle_expires_at = last_activity_at + timedelta(seconds=idle_timeout)
absolute_expires_at = created_at + timedelta(seconds=absolute_timeout)
if self.is_superadmin:
# Superadmin: absolute timeout only
return (
self.status == SessionStatus.ACTIVE
and now < absolute_expires_at
and self.deleted_at is None
)
# User: idle + absolute timeout
idle_timeout = current_app.config.get("SESSION_IDLE_TIMEOUT", 900)
idle_expires_at = last_activity_at + timedelta(seconds=idle_timeout)
return (
self.status == SessionStatus.ACTIVE
and now < idle_expires_at
@@ -83,6 +117,8 @@ class Session(BaseModel):
capped so that the session never exceeds the absolute lifetime
(``created_at + absolute timeout``).
Superadmin sessions only update last_activity_at (no sliding window).
Args:
duration_seconds: Override for the idle timeout. When *None*
(the common case), the value is read from
@@ -90,6 +126,12 @@ class Session(BaseModel):
"""
now = datetime.now(timezone.utc)
if self.is_superadmin:
# Superadmin: just bump last_activity_at, no sliding window
self.last_activity_at = now
db.session.commit()
return
if duration_seconds is None:
duration_seconds = current_app.config.get("SESSION_IDLE_TIMEOUT", 900)