User Management

from framefox.core.database.entity import Entity
from framefox.core.database.column import Column
from typing import List

class User(Entity):
    __tablename__ = "users"
    
    id: int = Column(primary_key=True, auto_increment=True)
    email: str = Column(unique=True, nullable=False)
    password: str = Column(nullable=False)  # Hashed password
    name: str = Column(nullable=False)
    roles: List[str] = Column(type="json", default=["ROLE_USER"])
    is_active: bool = Column(default=True)
    created_at: str = Column(type="datetime", auto_now_add=True)
    updated_at: str = Column(type="datetime", auto_now=True)

# app/repository/user_repository.py (auto-generated)
from framefox.core.database.repository import Repository
from app.entity.user import User

class UserRepository(Repository[User]):
    def __init__(self):
        super().__init__(User)

from framefox.core.security.user.user_provider import UserProvider

class ProfileController:
    def __init__(self, user_provider: UserProvider):
        self.user_provider = user_provider
    
    async def show_profile(self, request: Request):
        # Get current user
        user = self.user_provider.get_current_user()
        
        if not user:
            return RedirectResponse(url="/login")
        
        return templates.TemplateResponse("profile.html", {
            "request": request,
            "user": user
        })

def get_current_user(self, user_class: Type[T] = None) -> Optional[T]:
    # Check session cache first
    user_id_cache = self.session.get("user_id")
    if user_id_cache:
        # Handle virtual OAuth users
        if str(user_id_cache).startswith("oauth_"):
            payload = self.token_storage.get_payload()
            if payload:
                return self._create_virtual_user_from_payload(payload)
        
        # Get from database
        firewall_name = "main"
        provider_info = self.entity_user_provider.get_repository_and_property(firewall_name)
        
        if provider_info:
            repository, _ = provider_info
            user = repository.find(user_id_cache)
            if user:
                return user
    
    # Check JWT token
    payload = self.token_storage.get_payload()
    if not payload:
        return None
    
    user_id = payload.get("sub")
    if not user_id:
        return None
    
    # Handle virtual OAuth user
    if str(user_id).startswith("oauth_"):
        return self._create_virtual_user_from_payload(payload)
    
    # Get from repository
    provider_info = self.entity_user_provider.get_repository_and_property(
        payload.get("firewallname", "main")
    )
    
    if provider_info:
        repository, _ = provider_info
        user = repository.find(user_id)
        if user:
            # Cache in session
            self.session.set("user_id", user_id)
            self.session.save()
        return user
    
    return None

class EntityUserProvider:
    def get_repository_and_property(self, firewall_name: str) -> Optional[Tuple[Any, str]]:
        """Get repository and property name for a firewall."""
        firewalls = self.settings.firewalls
        firewall_config = firewalls.get(firewall_name, {})
        provider_name = firewall_config.get("provider")
        
        if not provider_name:
            return None
        
        provider_config = self.settings.providers.get(provider_name, {})
        entity_class_path = provider_config.get("entity", {}).get("class")
        property_name = provider_config.get("entity", {}).get("property")
        
        if not entity_class_path or not property_name:
            return None
        
        # Load repository dynamically
        module_path, class_name = entity_class_path.rsplit(".", 1)
        repository_module_path = module_path.replace(".entity.", ".repository.") + "_repository"
        repository_class_name = f"{class_name}Repository"
        
        module = importlib.import_module(repository_module_path)
        repository_class = getattr(module, repository_class_name)
        repository_instance = repository_class()
        
        return repository_instance, property_name

security:
  providers:
    app_user_provider:
      entity:
        class: app.entity.user.User
        property: email  # Property to use for user lookup
    
    api_user_provider:
      entity:
        class: app.entity.api_user.ApiUser
        property: username

class UserBadge:
    def __init__(self, user_identifier: str, user_identifier_property: str = "email"):
        self.user_identifier = user_identifier
        self.user_identifier_property = user_identifier_property
    
    async def get_user(self, repository):
        """Retrieve user from repository by identifier."""
        user = repository.find_by({
            self.user_identifier_property: self.user_identifier
        })
        
        if user:
            return user[0]
        return None

from framefox.core.security.passport.user_badge import UserBadge

# Create user badge with email
user_badge = UserBadge(
    user_identifier="user@example.com",
    user_identifier_property="email"
)

# Or with username
user_badge = UserBadge(
    user_identifier="johndoe",
    user_identifier_property="username"
)

from framefox.core.security.password.password_hasher import PasswordHasher

class PasswordHasher:
    def __init__(self):
        self.pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
    
    def hash(self, password: str) -> str:
        """Hash a password using bcrypt."""
        return self.pwd_context.hash(password)
    
    def verify(self, raw_password: str, hashed_password: str) -> bool:
        """Verify a password against its hash."""
        return self.pwd_context.verify(raw_password, hashed_password)

from framefox.core.security.password.password_hasher import PasswordHasher

hasher = PasswordHasher()

# Hash password before storing
hashed_password = hasher.hash("user_password")

# Store in database
user = User(
    email="user@example.com",
    password=hashed_password,
    name="John Doe",
    roles=["ROLE_USER"]
)
repository.save(user)

# During authentication
user = repository.find_by({"email": "user@example.com"})[0]

hasher = PasswordHasher()
if hasher.verify(submitted_password, user.password):
    # Password is correct
    pass
else:
    # Password is incorrect
    pass

class User(Entity):
    roles: List[str] = Column(type="json", default=["ROLE_USER"])

security:
  access_control:
    # Admin routes
    - { path: ^/admin, roles: [ROLE_ADMIN] }
    
    # API routes (authenticated users)
    - { path: ^/api, roles: [ROLE_USER] }
    
    # Moderator routes
    - { path: ^/moderate, roles: [ROLE_MODERATOR, ROLE_ADMIN] }
    
    # Public routes
    - { path: ^/public, roles: IS_AUTHENTICATED_ANONYMOUSLY }
    - { path: ^/about, roles: IS_AUTHENTICATED_ANONYMOUSLY }
  
  # Default policy for unmatched routes
  default_access_policy: deny  # or 'allow'

class AccessManager:
    def get_required_roles(self, path: str) -> List[str]:
        """Get required roles for a path."""
        if not self.settings.access_control:
            return []
        
        for rule in self.settings.access_control:
            pattern = rule.get("path")
            roles = rule.get("roles", [])
            
            if isinstance(roles, str):
                roles = [roles]
            
            if re.match(pattern, path):
                return roles
        
        # Apply default policy
        default_policy = self.settings.config.get("security", {}).get(
            "default_access_policy", "allow"
        )
        
        if default_policy == "deny":
            return ["ROLE_ADMIN"]
        else:
            return []
    
    def is_allowed(self, user_roles: List[str], required_roles: List[str]) -> bool:
        """Check if user has required roles."""
        if "IS_AUTHENTICATED_ANONYMOUSLY" in required_roles:
            return True
        
        if not required_roles:
            return True
        
        if not user_roles:
            return False
        
        # User needs at least one of the required roles
        return any(role in user_roles for role in required_roles)

from framefox.core.security.user.user_provider import UserProvider
from fastapi import HTTPException

class AdminController:
    def __init__(self, user_provider: UserProvider):
        self.user_provider = user_provider
    
    async def admin_dashboard(self, request: Request):
        user = self.user_provider.get_current_user()
        
        if not user or "ROLE_ADMIN" not in user.roles:
            raise HTTPException(status_code=403, detail="Access denied")
        
        return templates.TemplateResponse("admin/dashboard.html", {
            "request": request,
            "user": user
        })

<!-- Check if user has specific role -->
{% if user and 'ROLE_ADMIN' in user.roles %}
    <a href="/admin">Admin Panel</a>
{% endif %}

<!-- Check if user is authenticated -->
{% if user %}
    <p>Welcome, {{ user.name }}!</p>
    <a href="/logout">Logout</a>
{% else %}
    <a href="/login">Login</a>
{% endif %}

def _create_virtual_oauth_user(self, default_roles: list = None):
    """Create a virtual user for OAuth authentication."""
    from types import SimpleNamespace
    
    if default_roles is None:
        default_roles = ["ROLE_USER"]
    
    virtual_user = SimpleNamespace()
    virtual_user.id = f"oauth_{hash(self.user_badge.user_identifier)}"
    virtual_user.email = self.user_badge.user_identifier
    virtual_user.name = self.user_badge.user_identifier.split("@")[0]
    virtual_user.roles = default_roles
    virtual_user.provider = "oauth"
    virtual_user.is_virtual = True
    
    return virtual_user

user = user_provider.get_current_user()

if hasattr(user, "is_virtual") and user.is_virtual:
    # This is a virtual OAuth user
    print(f"OAuth user from {user.provider}")
else:
    # This is a database user
    print(f"Database user with ID {user.id}")

from framefox.core.security.password.password_hasher import PasswordHasher
from app.repository.user_repository import UserRepository
from app.entity.user import User

class RegistrationController:
    def __init__(
        self,
        hasher: PasswordHasher,
        user_repository: UserRepository
    ):
        self.hasher = hasher
        self.user_repository = user_repository
    
    async def register(self, request: Request):
        form_data = await request.form()
        
        email = form_data.get("email")
        password = form_data.get("password")
        name = form_data.get("name")
        
        # Check if user exists
        existing = self.user_repository.find_by({"email": email})
        if existing:
            return templates.TemplateResponse("register.html", {
                "request": request,
                "error": "Email already registered"
            })
        
        # Hash password
        hashed_password = self.hasher.hash(password)
        
        # Create user
        user = User(
            email=email,
            password=hashed_password,
            name=name,
            roles=["ROLE_USER"]
        )
        
        # Save to database
        self.user_repository.save(user)
        
        return RedirectResponse(url="/login", status_code=303)

<form method="POST" action="/register">
    <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
    
    <label for="name">Name:</label>
    <input type="text" name="name" id="name" required>
    
    <label for="email">Email:</label>
    <input type="email" name="email" id="email" required>
    
    <label for="password">Password:</label>
    <input type="password" name="password" id="password" required minlength="8">
    
    <label for="confirm_password">Confirm Password:</label>
    <input type="password" name="confirm_password" id="confirm_password" required>
    
    <button type="submit">Register</button>
</form>

class ProfileController:
    def __init__(
        self,
        user_provider: UserProvider,
        user_repository: UserRepository
    ):
        self.user_provider = user_provider
        self.user_repository = user_repository
    
    async def update_profile(self, request: Request):
        user = self.user_provider.get_current_user()
        if not user:
            return RedirectResponse(url="/login")
        
        form_data = await request.form()
        
        # Update user fields
        user.name = form_data.get("name", user.name)
        
        # Save changes
        self.user_repository.update(user)
        
        return RedirectResponse(url="/profile", status_code=303)

async def change_password(self, request: Request):
    user = self.user_provider.get_current_user()
    if not user:
        return RedirectResponse(url="/login")
    
    form_data = await request.form()
    current_password = form_data.get("current_password")
    new_password = form_data.get("new_password")
    
    # Verify current password
    hasher = PasswordHasher()
    if not hasher.verify(current_password, user.password):
        return templates.TemplateResponse("change_password.html", {
            "request": request,
            "error": "Current password is incorrect"
        })
    
    # Hash and save new password
    user.password = hasher.hash(new_password)
    self.user_repository.update(user)
    
    return RedirectResponse(url="/profile?success=1", status_code=303)

# Never store plain text passwords
user.password = hasher.hash(password)  # Good
user.password = password  # Bad

import re

def is_strong_password(password: str) -> bool:
    if len(password) < 8:
        return False
    if not re.search(r"[A-Z]", password):
        return False
    if not re.search(r"[a-z]", password):
        return False
    if not re.search(r"[0-9]", password):
        return False
    return True

existing = user_repository.find_by({"email": email})
if existing:
    raise ValueError("Email already registered")

class User(Entity):
    is_active: bool = Column(default=False)
    activation_token: str = Column(nullable=True)

ROLE_HIERARCHY = {
    "ROLE_ADMIN": ["ROLE_MODERATOR", "ROLE_USER"],
    "ROLE_MODERATOR": ["ROLE_USER"],
    "ROLE_USER": []
}