566 lines
20 KiB
Python
566 lines
20 KiB
Python
"""
|
|
Models for User Service
|
|
"""
|
|
|
|
import uuid
|
|
from pydantic import BaseModel, Field, EmailStr
|
|
from typing import List, Dict, Any, Optional
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
from modules.shared.attributeUtils import register_model_labels, AttributeDefinition, ModelMixin
|
|
from modules.shared.timezoneUtils import get_utc_timestamp
|
|
|
|
class AuthAuthority(str, Enum):
|
|
"""Authentication authority enum"""
|
|
LOCAL = "local"
|
|
GOOGLE = "google"
|
|
MSFT = "msft"
|
|
|
|
class UserPrivilege(str, Enum):
|
|
"""User privilege levels"""
|
|
SYSADMIN = "sysadmin"
|
|
ADMIN = "admin"
|
|
USER = "user"
|
|
|
|
class ConnectionStatus(str, Enum):
|
|
"""Connection status"""
|
|
ACTIVE = "active"
|
|
EXPIRED = "expired"
|
|
REVOKED = "revoked"
|
|
PENDING = "pending"
|
|
|
|
class TokenStatus(str, Enum):
|
|
"""Status of an issued gateway JWT access token"""
|
|
ACTIVE = "active"
|
|
REVOKED = "revoked"
|
|
|
|
class Mandate(BaseModel, ModelMixin):
|
|
"""Data model for a mandate"""
|
|
id: str = Field(
|
|
default_factory=lambda: str(uuid.uuid4()),
|
|
description="Unique ID of the mandate",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
name: str = Field(
|
|
description="Name of the mandate",
|
|
frontend_type="text",
|
|
frontend_readonly=False,
|
|
frontend_required=True
|
|
)
|
|
language: str = Field(
|
|
default="en",
|
|
description="Default language of the mandate",
|
|
frontend_type="select",
|
|
frontend_readonly=False,
|
|
frontend_required=True,
|
|
frontend_options=[
|
|
{"value": "de", "label": {"en": "Deutsch", "fr": "Allemand"}},
|
|
{"value": "en", "label": {"en": "English", "fr": "Anglais"}},
|
|
{"value": "fr", "label": {"en": "Français", "fr": "Français"}},
|
|
{"value": "it", "label": {"en": "Italiano", "fr": "Italien"}}
|
|
]
|
|
)
|
|
enabled: bool = Field(
|
|
default=True,
|
|
description="Indicates whether the mandate is enabled",
|
|
frontend_type="checkbox",
|
|
frontend_readonly=False,
|
|
frontend_required=False
|
|
)
|
|
|
|
# Register labels for Mandate
|
|
register_model_labels(
|
|
"Mandate",
|
|
{"en": "Mandate", "fr": "Mandat"},
|
|
{
|
|
"id": {"en": "ID", "fr": "ID"},
|
|
"name": {"en": "Name", "fr": "Nom"},
|
|
"language": {"en": "Language", "fr": "Langue"},
|
|
"enabled": {"en": "Enabled", "fr": "Activé"}
|
|
}
|
|
)
|
|
|
|
class UserConnection(BaseModel, ModelMixin):
|
|
"""Data model for a user's connection to an external service"""
|
|
id: str = Field(
|
|
default_factory=lambda: str(uuid.uuid4()),
|
|
description="Unique ID of the connection",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
userId: str = Field(
|
|
description="ID of the user this connection belongs to",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
authority: AuthAuthority = Field(
|
|
description="Authentication authority",
|
|
frontend_type="select",
|
|
frontend_readonly=True,
|
|
frontend_required=False,
|
|
frontend_options=[
|
|
{"value": "local", "label": {"en": "Local", "fr": "Local"}},
|
|
{"value": "google", "label": {"en": "Google", "fr": "Google"}},
|
|
{"value": "msft", "label": {"en": "Microsoft", "fr": "Microsoft"}}
|
|
]
|
|
)
|
|
externalId: str = Field(
|
|
description="User ID in the external system",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
externalUsername: str = Field(
|
|
description="Username in the external system",
|
|
frontend_type="text",
|
|
frontend_readonly=False,
|
|
frontend_required=False
|
|
)
|
|
externalEmail: Optional[EmailStr] = Field(
|
|
None,
|
|
description="Email in the external system",
|
|
frontend_type="email",
|
|
frontend_readonly=False,
|
|
frontend_required=False
|
|
)
|
|
status: ConnectionStatus = Field(
|
|
default=ConnectionStatus.ACTIVE,
|
|
description="Connection status",
|
|
frontend_type="select",
|
|
frontend_readonly=False,
|
|
frontend_required=False,
|
|
frontend_options=[
|
|
{"value": "active", "label": {"en": "Active", "fr": "Actif"}},
|
|
{"value": "inactive", "label": {"en": "Inactive", "fr": "Inactif"}},
|
|
{"value": "expired", "label": {"en": "Expired", "fr": "Expiré"}},
|
|
{"value": "pending", "label": {"en": "Pending", "fr": "En attente"}}
|
|
]
|
|
)
|
|
connectedAt: float = Field(
|
|
default_factory=get_utc_timestamp,
|
|
description="When the connection was established (UTC timestamp in seconds)",
|
|
frontend_type="timestamp",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
lastChecked: float = Field(
|
|
default_factory=get_utc_timestamp,
|
|
description="When the connection was last verified (UTC timestamp in seconds)",
|
|
frontend_type="timestamp",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
expiresAt: Optional[float] = Field(
|
|
None,
|
|
description="When the connection expires (UTC timestamp in seconds)",
|
|
frontend_type="timestamp",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
tokenStatus: Optional[str] = Field(
|
|
None,
|
|
description="Current token status: active, expired, none",
|
|
frontend_type="select",
|
|
frontend_readonly=True,
|
|
frontend_required=False,
|
|
frontend_options=[
|
|
{"value": "active", "label": {"en": "Active", "fr": "Actif"}},
|
|
{"value": "expired", "label": {"en": "Expired", "fr": "Expiré"}},
|
|
{"value": "none", "label": {"en": "None", "fr": "Aucun"}}
|
|
]
|
|
)
|
|
tokenExpiresAt: Optional[float] = Field(
|
|
None,
|
|
description="When the current token expires (UTC timestamp in seconds)",
|
|
frontend_type="timestamp",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
|
|
# Register labels for UserConnection
|
|
register_model_labels(
|
|
"UserConnection",
|
|
{"en": "User Connection", "fr": "Connexion utilisateur"},
|
|
{
|
|
"id": {"en": "ID", "fr": "ID"},
|
|
"userId": {"en": "User ID", "fr": "ID utilisateur"},
|
|
"authority": {"en": "Authority", "fr": "Autorité"},
|
|
"externalId": {"en": "External ID", "fr": "ID externe"},
|
|
"externalUsername": {"en": "External Username", "fr": "Nom d'utilisateur externe"},
|
|
"externalEmail": {"en": "External Email", "fr": "Email externe"},
|
|
"status": {"en": "Status", "fr": "Statut"},
|
|
"connectedAt": {"en": "Connected At", "fr": "Connecté le"},
|
|
"lastChecked": {"en": "Last Checked", "fr": "Dernière vérification"},
|
|
"expiresAt": {"en": "Expires At", "fr": "Expire le"},
|
|
"tokenStatus": {"en": "Connection Status", "fr": "Statut de connexion"},
|
|
"tokenExpiresAt": {"en": "Expires At", "fr": "Expire le"}
|
|
}
|
|
)
|
|
|
|
|
|
|
|
class User(BaseModel, ModelMixin):
|
|
"""Data model for a user"""
|
|
id: str = Field(
|
|
default_factory=lambda: str(uuid.uuid4()),
|
|
description="Unique ID of the user",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
username: str = Field(
|
|
description="Username for login",
|
|
frontend_type="text",
|
|
frontend_readonly=False,
|
|
frontend_required=True
|
|
)
|
|
email: Optional[EmailStr] = Field(
|
|
None,
|
|
description="Email address of the user",
|
|
frontend_type="email",
|
|
frontend_readonly=False,
|
|
frontend_required=True
|
|
)
|
|
fullName: Optional[str] = Field(
|
|
None,
|
|
description="Full name of the user",
|
|
frontend_type="text",
|
|
frontend_readonly=False,
|
|
frontend_required=False
|
|
)
|
|
language: str = Field(
|
|
default="en",
|
|
description="Preferred language of the user",
|
|
frontend_type="select",
|
|
frontend_readonly=False,
|
|
frontend_required=True,
|
|
frontend_options=[
|
|
{"value": "de", "label": {"en": "Deutsch", "fr": "Allemand"}},
|
|
{"value": "en", "label": {"en": "English", "fr": "Anglais"}},
|
|
{"value": "fr", "label": {"en": "Français", "fr": "Français"}},
|
|
{"value": "it", "label": {"en": "Italiano", "fr": "Italien"}}
|
|
]
|
|
)
|
|
enabled: bool = Field(
|
|
default=True,
|
|
description="Indicates whether the user is enabled",
|
|
frontend_type="checkbox",
|
|
frontend_readonly=False,
|
|
frontend_required=False
|
|
)
|
|
privilege: UserPrivilege = Field(
|
|
default=UserPrivilege.USER,
|
|
description="Permission level",
|
|
frontend_type="select",
|
|
frontend_readonly=False,
|
|
frontend_required=True,
|
|
frontend_options=[
|
|
{"value": "user", "label": {"en": "User", "fr": "Utilisateur"}},
|
|
{"value": "admin", "label": {"en": "Admin", "fr": "Administrateur"}},
|
|
{"value": "sysadmin", "label": {"en": "SysAdmin", "fr": "Administrateur système"}}
|
|
]
|
|
)
|
|
authenticationAuthority: AuthAuthority = Field(
|
|
default=AuthAuthority.LOCAL,
|
|
description="Primary authentication authority",
|
|
frontend_type="select",
|
|
frontend_readonly=True,
|
|
frontend_required=False,
|
|
frontend_options=[
|
|
{"value": "local", "label": {"en": "Local", "fr": "Local"}},
|
|
{"value": "google", "label": {"en": "Google", "fr": "Google"}},
|
|
{"value": "msft", "label": {"en": "Microsoft", "fr": "Microsoft"}}
|
|
]
|
|
)
|
|
mandateId: Optional[str] = Field(
|
|
None,
|
|
description="ID of the mandate this user belongs to",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
|
|
# Register labels for User
|
|
register_model_labels(
|
|
"User",
|
|
{"en": "User", "fr": "Utilisateur"},
|
|
{
|
|
"id": {"en": "ID", "fr": "ID"},
|
|
"username": {"en": "Username", "fr": "Nom d'utilisateur"},
|
|
"email": {"en": "Email", "fr": "Email"},
|
|
"fullName": {"en": "Full Name", "fr": "Nom complet"},
|
|
"language": {"en": "Language", "fr": "Langue"},
|
|
"enabled": {"en": "Enabled", "fr": "Activé"},
|
|
"privilege": {"en": "Privilege", "fr": "Privilège"},
|
|
"authenticationAuthority": {"en": "Auth Authority", "fr": "Autorité d'authentification"},
|
|
"mandateId": {"en": "Mandate ID", "fr": "ID de mandat"}
|
|
}
|
|
)
|
|
|
|
class UserInDB(User):
|
|
"""Extended user class with password hash"""
|
|
hashedPassword: Optional[str] = Field(None, description="Hash of the user password")
|
|
|
|
# Register labels for UserInDB
|
|
register_model_labels(
|
|
"UserInDB",
|
|
{"en": "User Access", "fr": "Accès de l'utilisateur"},
|
|
{
|
|
"hashedPassword": {"en": "Password hash", "fr": "Hachage de mot de passe"}
|
|
}
|
|
)
|
|
|
|
# Token Models
|
|
class Token(BaseModel, ModelMixin):
|
|
"""Token model for all authentication types"""
|
|
id: Optional[str] = None
|
|
userId: str
|
|
authority: AuthAuthority
|
|
connectionId: Optional[str] = Field(None, description="ID of the connection this token belongs to")
|
|
tokenAccess: str
|
|
tokenType: str = "bearer"
|
|
expiresAt: float = Field(description="When the token expires (UTC timestamp in seconds)")
|
|
tokenRefresh: Optional[str] = None
|
|
createdAt: Optional[float] = Field(None, description="When the token was created (UTC timestamp in seconds)")
|
|
# Revocation and session tracking (for LOCAL gateway JWTs)
|
|
status: TokenStatus = Field(default=TokenStatus.ACTIVE, description="Token status: active/revoked")
|
|
revokedAt: Optional[float] = Field(None, description="When the token was revoked (UTC timestamp in seconds)")
|
|
revokedBy: Optional[str] = Field(None, description="User ID who revoked the token (admin/self)")
|
|
reason: Optional[str] = Field(None, description="Optional revocation reason")
|
|
sessionId: Optional[str] = Field(None, description="Logical session grouping for logout revocation")
|
|
mandateId: Optional[str] = Field(None, description="Mandate ID for tenant scoping of the token")
|
|
|
|
class Config:
|
|
useEnumValues = True
|
|
|
|
# Register labels for Token
|
|
register_model_labels(
|
|
"Token",
|
|
{"en": "Token", "fr": "Jeton"},
|
|
{
|
|
"id": {"en": "ID", "fr": "ID"},
|
|
"userId": {"en": "User ID", "fr": "ID utilisateur"},
|
|
"authority": {"en": "Authority", "fr": "Autorité"},
|
|
"connectionId": {"en": "Connection ID", "fr": "ID de connexion"},
|
|
"tokenAccess": {"en": "Access Token", "fr": "Jeton d'accès"},
|
|
"tokenType": {"en": "Token Type", "fr": "Type de jeton"},
|
|
"expiresAt": {"en": "Expires At", "fr": "Expire le"},
|
|
"tokenRefresh": {"en": "Refresh Token", "fr": "Jeton de rafraîchissement"},
|
|
"createdAt": {"en": "Created At", "fr": "Créé le"},
|
|
"status": {"en": "Status", "fr": "Statut"},
|
|
"revokedAt": {"en": "Revoked At", "fr": "Révoqué le"},
|
|
"revokedBy": {"en": "Revoked By", "fr": "Révoqué par"},
|
|
"reason": {"en": "Reason", "fr": "Raison"},
|
|
"sessionId": {"en": "Session ID", "fr": "ID de session"},
|
|
"mandateId": {"en": "Mandate ID", "fr": "ID de mandat"}
|
|
}
|
|
)
|
|
|
|
class LocalToken(Token):
|
|
"""Local authentication token model"""
|
|
pass
|
|
|
|
class GoogleToken(Token):
|
|
"""Google OAuth token model"""
|
|
pass
|
|
|
|
class MsftToken(Token):
|
|
"""Microsoft OAuth token model"""
|
|
pass
|
|
|
|
class AuthEvent(BaseModel, ModelMixin):
|
|
"""Data model for authentication events"""
|
|
id: str = Field(
|
|
default_factory=lambda: str(uuid.uuid4()),
|
|
description="Unique ID of the auth event",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
userId: str = Field(
|
|
description="ID of the user this event belongs to",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=True
|
|
)
|
|
eventType: str = Field(
|
|
description="Type of authentication event (e.g., 'login', 'logout', 'token_refresh')",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=True
|
|
)
|
|
timestamp: float = Field(
|
|
default_factory=get_utc_timestamp,
|
|
description="Unix timestamp when the event occurred",
|
|
frontend_type="datetime",
|
|
frontend_readonly=True,
|
|
frontend_required=True
|
|
)
|
|
ipAddress: Optional[str] = Field(
|
|
default=None,
|
|
description="IP address from which the event originated",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
userAgent: Optional[str] = Field(
|
|
default=None,
|
|
description="User agent string from the request",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
success: bool = Field(
|
|
default=True,
|
|
description="Whether the authentication event was successful",
|
|
frontend_type="boolean",
|
|
frontend_readonly=True,
|
|
frontend_required=True
|
|
)
|
|
details: Optional[str] = Field(
|
|
default=None,
|
|
description="Additional details about the event",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
|
|
# Register labels for AuthEvent
|
|
register_model_labels(
|
|
"AuthEvent",
|
|
{"en": "Authentication Event", "fr": "Événement d'authentification"},
|
|
{
|
|
"id": {"en": "ID", "fr": "ID"},
|
|
"userId": {"en": "User ID", "fr": "ID utilisateur"},
|
|
"eventType": {"en": "Event Type", "fr": "Type d'événement"},
|
|
"timestamp": {"en": "Timestamp", "fr": "Horodatage"},
|
|
"ipAddress": {"en": "IP Address", "fr": "Adresse IP"},
|
|
"userAgent": {"en": "User Agent", "fr": "Agent utilisateur"},
|
|
"success": {"en": "Success", "fr": "Succès"},
|
|
"details": {"en": "Details", "fr": "Détails"}
|
|
}
|
|
)
|
|
|
|
class DataNeutraliserConfig(BaseModel, ModelMixin):
|
|
"""Data model for data neutralization configuration"""
|
|
id: str = Field(
|
|
default_factory=lambda: str(uuid.uuid4()),
|
|
description="Unique ID of the configuration",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
mandateId: str = Field(
|
|
description="ID of the mandate this configuration belongs to",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=True
|
|
)
|
|
userId: str = Field(
|
|
description="ID of the user who created this configuration",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=True
|
|
)
|
|
enabled: bool = Field(
|
|
default=True,
|
|
description="Whether data neutralization is enabled",
|
|
frontend_type="checkbox",
|
|
frontend_readonly=False,
|
|
frontend_required=False
|
|
)
|
|
namesToParse: str = Field(
|
|
default="",
|
|
description="Multiline list of names to parse for neutralization",
|
|
frontend_type="textarea",
|
|
frontend_readonly=False,
|
|
frontend_required=False
|
|
)
|
|
sharepointSourcePath: str = Field(
|
|
default="",
|
|
description="SharePoint path to read files for neutralization",
|
|
frontend_type="text",
|
|
frontend_readonly=False,
|
|
frontend_required=False
|
|
)
|
|
sharepointTargetPath: str = Field(
|
|
default="",
|
|
description="SharePoint path to store neutralized files",
|
|
frontend_type="text",
|
|
frontend_readonly=False,
|
|
frontend_required=False
|
|
)
|
|
|
|
# Register labels for DataNeutraliserConfig
|
|
register_model_labels(
|
|
"DataNeutraliserConfig",
|
|
{"en": "Data Neutralization Config", "fr": "Configuration de neutralisation des données"},
|
|
{
|
|
"id": {"en": "ID", "fr": "ID"},
|
|
"mandateId": {"en": "Mandate ID", "fr": "ID de mandat"},
|
|
"userId": {"en": "User ID", "fr": "ID utilisateur"},
|
|
"enabled": {"en": "Enabled", "fr": "Activé"},
|
|
"namesToParse": {"en": "Names to Parse", "fr": "Noms à analyser"},
|
|
"sharepointSourcePath": {"en": "Source Path", "fr": "Chemin source"},
|
|
"sharepointTargetPath": {"en": "Target Path", "fr": "Chemin cible"}
|
|
}
|
|
)
|
|
|
|
class DataNeutralizerAttributes(BaseModel, ModelMixin):
|
|
"""Data model for neutralized data attributes mapping"""
|
|
id: str = Field(
|
|
default_factory=lambda: str(uuid.uuid4()),
|
|
description="Unique ID of the attribute mapping (used as UID in neutralized files)",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
mandateId: str = Field(
|
|
description="ID of the mandate this attribute belongs to",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=True
|
|
)
|
|
userId: str = Field(
|
|
description="ID of the user who created this attribute",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=True
|
|
)
|
|
originalText: str = Field(
|
|
description="Original text that was neutralized",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=True
|
|
)
|
|
fileId: Optional[str] = Field(
|
|
default=None,
|
|
description="ID of the file this attribute belongs to",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=False
|
|
)
|
|
patternType: str = Field(
|
|
description="Type of pattern that matched (email, phone, name, etc.)",
|
|
frontend_type="text",
|
|
frontend_readonly=True,
|
|
frontend_required=True
|
|
)
|
|
|
|
# Register labels for DataNeutralizerAttributes
|
|
register_model_labels(
|
|
"DataNeutralizerAttributes",
|
|
{"en": "Neutralized Data Attribute", "fr": "Attribut de données neutralisées"},
|
|
{
|
|
"id": {"en": "ID", "fr": "ID"},
|
|
"mandateId": {"en": "Mandate ID", "fr": "ID de mandat"},
|
|
"userId": {"en": "User ID", "fr": "ID utilisateur"},
|
|
"originalText": {"en": "Original Text", "fr": "Texte original"},
|
|
"fileId": {"en": "File ID", "fr": "ID de fichier"},
|
|
"patternType": {"en": "Pattern Type", "fr": "Type de modèle"}
|
|
}
|
|
)
|