328 lines
12 KiB
Python
328 lines
12 KiB
Python
"""
|
|
Database models für OIDC Server
|
|
SQLAlchemy ORM Models für User, Authorization Codes und Access Tokens
|
|
"""
|
|
|
|
from datetime import datetime, timedelta
|
|
import json
|
|
|
|
# Import from app.core instead of creating new db instance
|
|
from app.core.database import db
|
|
from app.core.security import hash_password, verify_password, generate_secure_token
|
|
|
|
|
|
class User(db.Model):
|
|
"""User Model - Speichert alle User-Informationen"""
|
|
__tablename__ = 'users'
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
username = db.Column(db.String(80), unique=True, nullable=False, index=True)
|
|
password_hash = db.Column(db.String(128), nullable=False)
|
|
email = db.Column(db.String(120), unique=True, nullable=False)
|
|
name = db.Column(db.String(120), nullable=False)
|
|
preferred_username = db.Column(db.String(80), nullable=False)
|
|
is_admin = db.Column(db.Boolean, default=False)
|
|
role = db.Column(db.String(50), default='user') # Rolle: user, admin, moderator, readonly, etc.
|
|
permissions = db.Column(db.Text, default='[]') # JSON Array von Permissions
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
|
|
def __repr__(self):
|
|
return f'<User {self.username}>'
|
|
|
|
@property
|
|
def sub(self):
|
|
"""OIDC Subject Identifier - eindeutige User-ID"""
|
|
return f"user-{self.id}"
|
|
|
|
def set_password(self, password):
|
|
"""Hasht das Passwort mit bcrypt (uses app.core.security)"""
|
|
self.password_hash = hash_password(password)
|
|
|
|
def check_password(self, password):
|
|
"""Verifiziert das Passwort gegen den gespeicherten Hash (uses app.core.security)"""
|
|
return verify_password(password, self.password_hash)
|
|
|
|
def get_permissions(self):
|
|
"""Gibt die Permissions als Python-Liste zurück"""
|
|
try:
|
|
return json.loads(self.permissions) if self.permissions else []
|
|
except (json.JSONDecodeError, TypeError):
|
|
return []
|
|
|
|
def set_permissions(self, permissions_list):
|
|
"""Setzt Permissions aus einer Python-Liste"""
|
|
self.permissions = json.dumps(permissions_list)
|
|
|
|
def add_permission(self, permission):
|
|
"""Fügt eine einzelne Permission hinzu"""
|
|
perms = self.get_permissions()
|
|
if permission not in perms:
|
|
perms.append(permission)
|
|
self.set_permissions(perms)
|
|
|
|
def remove_permission(self, permission):
|
|
"""Entfernt eine einzelne Permission"""
|
|
perms = self.get_permissions()
|
|
if permission in perms:
|
|
perms.remove(permission)
|
|
self.set_permissions(perms)
|
|
|
|
def has_permission(self, permission):
|
|
"""Prüft ob User eine bestimmte Permission hat"""
|
|
return permission in self.get_permissions()
|
|
|
|
def to_dict(self):
|
|
"""Konvertiert User zu Dictionary für Token/UserInfo"""
|
|
return {
|
|
'sub': self.sub,
|
|
'username': self.username,
|
|
'email': self.email,
|
|
'name': self.name,
|
|
'preferred_username': self.preferred_username,
|
|
'role': self.role,
|
|
'permissions': self.get_permissions()
|
|
}
|
|
|
|
|
|
class Client(db.Model):
|
|
"""OIDC Client Model"""
|
|
__tablename__ = 'oidc_clients'
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
client_id = db.Column(db.String(48), unique=True, nullable=False, index=True)
|
|
client_secret_hash = db.Column(db.String(128), nullable=True)
|
|
client_name = db.Column(db.String(120), nullable=False)
|
|
redirect_uris = db.Column(db.Text, nullable=False)
|
|
allowed_scopes = db.Column(db.Text, nullable=False, default='["openid", "profile", "email"]')
|
|
|
|
def __repr__(self):
|
|
return f'<Client {self.client_name}>'
|
|
|
|
def set_client_secret(self, client_secret):
|
|
"""Hashes the client secret with bcrypt (uses app.core.security)"""
|
|
self.client_secret_hash = hash_password(client_secret)
|
|
|
|
def check_client_secret(self, client_secret):
|
|
"""Verifies the client secret against the stored hash (uses app.core.security)"""
|
|
if not self.client_secret_hash:
|
|
return False
|
|
return verify_password(client_secret, self.client_secret_hash)
|
|
|
|
def get_redirect_uris(self):
|
|
"""Returns the redirect URIs as a Python list"""
|
|
try:
|
|
return json.loads(self.redirect_uris)
|
|
except (json.JSONDecodeError, TypeError):
|
|
return []
|
|
|
|
def get_allowed_scopes(self):
|
|
"""Returns the allowed scopes as a Python list"""
|
|
try:
|
|
return json.loads(self.allowed_scopes)
|
|
except (json.JSONDecodeError, TypeError):
|
|
return []
|
|
|
|
|
|
class AuthorizationCode(db.Model):
|
|
"""Authorization Code Model - Speichert ausgestellte Authorization Codes"""
|
|
__tablename__ = 'authorization_codes'
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
code = db.Column(db.String(128), unique=True, nullable=False, index=True)
|
|
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
|
|
client_id = db.Column(db.String(128), nullable=False)
|
|
redirect_uri = db.Column(db.String(512), nullable=False)
|
|
scope = db.Column(db.String(256), nullable=False)
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
expires_at = db.Column(db.DateTime, nullable=False)
|
|
used = db.Column(db.Boolean, default=False)
|
|
|
|
user = db.relationship('User', backref='authorization_codes')
|
|
|
|
def __repr__(self):
|
|
return f'<AuthorizationCode {self.code[:8]}...>'
|
|
|
|
@staticmethod
|
|
def create(user_id, client_id, redirect_uri, scope, ttl_seconds=600):
|
|
"""Erstellt einen neuen Authorization Code (uses app.core.security)"""
|
|
code = generate_secure_token(32)
|
|
expires_at = datetime.utcnow() + timedelta(seconds=ttl_seconds)
|
|
|
|
auth_code = AuthorizationCode(
|
|
code=code,
|
|
user_id=user_id,
|
|
client_id=client_id,
|
|
redirect_uri=redirect_uri,
|
|
scope=scope,
|
|
expires_at=expires_at
|
|
)
|
|
return auth_code
|
|
|
|
def is_valid(self):
|
|
"""Prüft ob der Code noch gültig ist"""
|
|
return not self.used and datetime.utcnow() < self.expires_at
|
|
|
|
def mark_used(self):
|
|
"""Markiert den Code als verwendet"""
|
|
self.used = True
|
|
|
|
|
|
class AccessToken(db.Model):
|
|
"""Access Token Model - Speichert ausgestellte Access Tokens"""
|
|
__tablename__ = 'access_tokens'
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
token = db.Column(db.String(128), unique=True, nullable=False, index=True)
|
|
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
|
|
client_id = db.Column(db.String(128), nullable=True, index=True) # Which client this token was issued for
|
|
scope = db.Column(db.String(256), nullable=False)
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
expires_at = db.Column(db.DateTime, nullable=False)
|
|
revoked = db.Column(db.Boolean, default=False)
|
|
|
|
user = db.relationship('User', backref='access_tokens')
|
|
|
|
def __repr__(self):
|
|
return f'<AccessToken {self.token[:8]}...>'
|
|
|
|
@staticmethod
|
|
def create(user_id, scope, client_id=None, ttl_seconds=3600):
|
|
"""Erstellt einen neuen Access Token (uses app.core.security)"""
|
|
token = generate_secure_token(32)
|
|
expires_at = datetime.utcnow() + timedelta(seconds=ttl_seconds)
|
|
|
|
access_token = AccessToken(
|
|
token=token,
|
|
user_id=user_id,
|
|
client_id=client_id,
|
|
scope=scope,
|
|
expires_at=expires_at
|
|
)
|
|
return access_token
|
|
|
|
def is_valid(self):
|
|
"""Prüft ob der Token noch gültig ist"""
|
|
return not self.revoked and datetime.utcnow() < self.expires_at
|
|
|
|
def revoke(self):
|
|
"""Widerruft den Token"""
|
|
self.revoked = True
|
|
|
|
|
|
class AuditLog(db.Model):
|
|
"""
|
|
Audit Log Model - Trackt wichtige Events (Login, Admin Actions, etc.)
|
|
"""
|
|
__tablename__ = 'audit_logs'
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
timestamp = db.Column(db.DateTime, default=datetime.utcnow, nullable=False, index=True)
|
|
action = db.Column(db.String(100), nullable=False, index=True) # login_success, login_failed, user_created, etc.
|
|
username = db.Column(db.String(80), nullable=True, index=True) # Username (falls bekannt)
|
|
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True) # User ID (falls bekannt)
|
|
ip_address = db.Column(db.String(45), nullable=True) # IPv4 oder IPv6
|
|
user_agent = db.Column(db.Text, nullable=True) # Browser/Client Info
|
|
details = db.Column(db.Text, nullable=True) # Zusätzliche Details als JSON
|
|
|
|
user = db.relationship('User', backref='audit_logs')
|
|
|
|
def __repr__(self):
|
|
return f'<AuditLog {self.action} by {self.username} at {self.timestamp}>'
|
|
|
|
@classmethod
|
|
def log(cls, action, username=None, user_id=None, ip_address=None, user_agent=None, details=None):
|
|
"""
|
|
Helper Methode zum einfachen Erstellen von Audit Logs
|
|
|
|
Args:
|
|
action: Art der Aktion (z.B. "login_success", "user_created")
|
|
username: Username (optional)
|
|
user_id: User ID (optional)
|
|
ip_address: IP Adresse (optional)
|
|
user_agent: User Agent String (optional)
|
|
details: Zusätzliche Details als Dict (optional, wird zu JSON konvertiert)
|
|
"""
|
|
audit_entry = cls(
|
|
action=action,
|
|
username=username,
|
|
user_id=user_id,
|
|
ip_address=ip_address,
|
|
user_agent=user_agent,
|
|
details=json.dumps(details) if details else None
|
|
)
|
|
db.session.add(audit_entry)
|
|
db.session.commit()
|
|
return audit_entry
|
|
|
|
|
|
def seed_db():
|
|
"""Seeds the database with initial data."""
|
|
if User.query.count() == 0:
|
|
print("Seeding database with initial users...")
|
|
admin = User(
|
|
username='admin',
|
|
email='admin@homelab.local',
|
|
name='Admin User',
|
|
preferred_username='admin',
|
|
is_admin=True,
|
|
role='admin',
|
|
permissions=json.dumps(['read:data', 'write:data', 'manage:users', 'manage:settings'])
|
|
)
|
|
admin.set_password('admin123')
|
|
|
|
test = User(
|
|
username='test',
|
|
email='test@homelab.local',
|
|
name='Test User',
|
|
preferred_username='test',
|
|
role='user',
|
|
permissions=json.dumps(['read:data'])
|
|
)
|
|
test.set_password('test123')
|
|
|
|
db.session.add(admin)
|
|
db.session.add(test)
|
|
print("Test-User erstellt: admin/admin123, test/test123")
|
|
|
|
if Client.query.count() == 0:
|
|
print("Seeding database with initial client...")
|
|
client_secret = generate_secure_token(32)
|
|
default_client = Client(
|
|
client_id='test-client',
|
|
client_name='Default Test Client',
|
|
redirect_uris=json.dumps(['http://localhost:8080/callback']),
|
|
allowed_scopes=json.dumps(['openid', 'profile', 'email'])
|
|
)
|
|
default_client.set_client_secret(client_secret)
|
|
db.session.add(default_client)
|
|
print(f"Default client created. Client ID: test-client, Client Secret: {client_secret}")
|
|
|
|
db.session.commit()
|
|
|
|
|
|
def init_db(app):
|
|
"""
|
|
Initialisiert die Datenbank
|
|
Note: This is kept for backwards compatibility.
|
|
Prefer using app.core.database.init_db() instead.
|
|
"""
|
|
from app.core.database import init_db as core_init_db
|
|
core_init_db(app)
|
|
|
|
|
|
def cleanup_expired_tokens():
|
|
"""Löscht abgelaufene Authorization Codes und Access Tokens"""
|
|
now = datetime.utcnow()
|
|
|
|
# Abgelaufene Authorization Codes löschen
|
|
expired_codes = AuthorizationCode.query.filter(AuthorizationCode.expires_at < now).delete()
|
|
|
|
# Abgelaufene Access Tokens löschen
|
|
expired_tokens = AccessToken.query.filter(AccessToken.expires_at < now).delete()
|
|
|
|
db.session.commit()
|
|
|
|
return expired_codes, expired_tokens
|