Roles and Privileges Assessment and Recommendation¶

# API Key Model
class ApiKey(Base):
    scopes = Column(JSON)  # e.g., ["templates:read", "presentations:generate", "*"]
    is_active = Column(Boolean)
    expires_at = Column(DateTime)
    usage_count = Column(Integer)
    rate_limit_per_hour = Column(Integer)

# Scope checking
def has_scope(self, scope: str) -> bool:
    scopes_list = self.scopes if self.scopes else []
    return scope in scopes_list or "*" in scopes_list

# Usage
if not api_key.has_scope("templates:write"):
    raise HTTPException(403, "Missing scope")

# User Model (Database - Currently UNUSED)
class User(Base):
    __tablename__ = "app_users"
    email = Column(String(255), unique=True)
    attributes = Column(JSON)  # JSON blob for permissions
    active = Column(Boolean)

# Config File (Currently ACTIVE)
ALLOWED_USERS = {
    "user@example.com": {
        "id": -100,
        "attributes": {
            "checks": ["view", "admin"],
            "prompts": ["view", "admin"],
            "contexts": ["view", "admin"],
            "users": ["view", "admin"],
            "files": ["view", "admin"],
            "workflows": ["esg2", "poc_ideation"]
        }
    }
}

# Workflow-level access check
user_workflows = user_config.get('attributes', {}).get('workflows', [])
if workflow_folder not in user_workflows:
    return None

# Resource-level permissions (app/filemanager/workflow_files/permissions.py)
def check_template_permission(user, workflow_folder, action):
    user_permissions = get_user_workflow_permissions(user, workflow_folder)
    template_perms = user_permissions.get('templates', [])
    return action in template_perms

def convert_attributes_to_scopes(attributes: dict) -> List[str]:
    scopes = []

    # Convert workflow access
    for workflow in attributes.get('workflows', []):
        scopes.append(f"workflows:{workflow}:read")
        scopes.append(f"workflows:{workflow}:execute")

    # Convert resource permissions
    resource_map = {
        'checks': 'checks',
        'prompts': 'prompts',
        'contexts': 'contexts',
        'users': 'users',
        'files': 'templates'
    }

    for attr_key, scope_resource in resource_map.items():
        perms = attributes.get(attr_key, [])
        if 'view' in perms:
            scopes.append(f"{scope_resource}:read")
        if 'admin' in perms:
            scopes.append(f"{scope_resource}:write")
            scopes.append(f"{scope_resource}:delete")

    return scopes

# Base authorization interface
class BaseAuth:
    scopes: List[str] = []

    def has_scope(self, scope: str) -> bool:
        """Check if has specific scope or wildcard."""
        return scope in self.scopes or "*" in self.scopes

    def has_workflow_access(self, workflow: str, action: str = "read") -> bool:
        """Check workflow-specific access with fallback to global."""
        # 1. Try workflow-specific scope
        if self.has_scope(f"workflows:{workflow}:{action}"):
            return True
        # 2. Fall back to global workflow scope
        if self.has_scope(f"workflows:{action}"):
            return True
        # 3. Fall back to wildcard
        return self.has_scope("*")

    def has_resource_access(self, resource: str, action: str, workflow: str = None) -> bool:
        """Check resource access with optional workflow scoping."""
        # 1. Try workflow-specific resource scope
        if workflow and self.has_scope(f"{resource}:{workflow}:{action}"):
            return True
        # 2. Fall back to global resource scope
        if self.has_scope(f"{resource}:{action}"):
            return True
        # 3. Fall back to wildcard
        return self.has_scope("*")

# API Key Auth (already has scopes from database)
class ApiKeyAuth(BaseAuth):
    def __init__(self, api_key: ApiKey, user_id: str, user_email: str = None):
        self.api_key = api_key
        self.scopes = api_key.scopes or []
        # ... rest of init

# Session Data (add scopes from user)
class SessionData(BaseModel, BaseAuth):
    id: Optional[int] = None
    email: Optional[str] = None
    scopes: List[str] = []  # NEW: Load from database
    # ... rest of fields

# In auth router, when creating session
async def login(username: str, password: str):
    # Verify credentials
    user = db.query(User).filter(User.email == username).first()
    if not user or not verify_password(password, user.password):
        raise HTTPException(401, "Invalid credentials")

    # Create session with scopes
    session_data = SessionData(
        id=user.id,
        email=user.email,
        name=user.name,
        scopes=user.scopes or [],  # Load from database
        attributes=user.attributes or {}
    )

    # Store session
    await backend.create(session_id, session_data)

# In app/api/auth.py

def require_scope(*required_scopes: str):
    """Dependency that requires one or more scopes."""
    async def _verify(
        user: Union[SessionData, ApiKeyAuth] = Depends(verify_api_access)
    ) -> Union[SessionData, ApiKeyAuth]:
        # Check if user has ANY of the required scopes
        for scope in required_scopes:
            if user.has_scope(scope):
                return user

        raise HTTPException(
            status_code=403,
            detail=f"Missing required scope. Need one of: {', '.join(required_scopes)}"
        )

    return _verify

def require_workflow_access(workflow_param: str = "workflow_folder", action: str = "read"):
    """Dependency that checks workflow-specific access."""
    async def _verify(
        user: Union[SessionData, ApiKeyAuth] = Depends(verify_api_access),
        workflow: str = Path(..., alias=workflow_param)
    ) -> Union[SessionData, ApiKeyAuth]:
        if not user.has_workflow_access(workflow, action):
            raise HTTPException(
                status_code=403,
                detail=f"Missing workflow access: {workflow}:{action}"
            )
        return user

    return _verify

# API endpoint with scope check
@router.post("/api/presentations/generate")
async def generate_presentation(
    request: PresentationGenerateRequest,
    user: Union[SessionData, ApiKeyAuth] = Depends(require_scope("presentations:generate"))
):
    # User guaranteed to have presentations:generate scope
    pass

# Workflow-specific template endpoint
@router.get("/api/workflow-files/workflows/{workflow_folder}/templates")
async def list_workflow_templates(
    workflow_folder: str,
    user: Union[SessionData, ApiKeyAuth] = Depends(require_workflow_access(action="read"))
):
    # User guaranteed to have workflows:{workflow_folder}:read scope
    # Additional check for templates
    if not user.has_resource_access("templates", "read", workflow_folder):
        raise HTTPException(403, "Cannot read templates for this workflow")

    templates = await list_templates(workflow_folder, user)
    return templates

# Flexible scope checking
@router.post("/api/workflow-files/workflows/{workflow_folder}/templates")
async def upload_template(
    workflow_folder: str,
    file: UploadFile,
    user: Union[SessionData, ApiKeyAuth] = Depends(verify_api_access)
):
    # Check workflow-specific OR global template write scope
    if not user.has_resource_access("templates", "write", workflow_folder):
        raise HTTPException(
            403,
            f"Missing scope: templates:{workflow_folder}:write or templates:write"
        )

    template = await upload_template(workflow_folder, file, user)
    return template

# app/filemanager/workflow_files/permissions.py - SIMPLIFIED

def check_template_permission(user: Union[SessionData, ApiKeyAuth],
                              workflow_folder: str,
                              action: str) -> bool:
    """Check if user can perform template action."""
    action_map = {'view': 'read', 'upload': 'write', 'delete': 'delete'}
    scope_action = action_map.get(action, action)
    return user.has_resource_access("templates", scope_action, workflow_folder)

def check_output_permission(user: Union[SessionData, ApiKeyAuth],
                           workflow_folder: str,
                           action: str) -> bool:
    """Check if user can perform output action."""
    action_map = {'view': 'read', 'download': 'read', 'delete': 'delete'}
    scope_action = action_map.get(action, action)
    return user.has_resource_access("outputs", scope_action, workflow_folder)

# Remove get_user_workflow_permissions - no longer needed
# Remove get_user_accessible_workflows - replaced by scope checking

# cli/commands/user.py

SCOPE_PRESETS = {
    'admin': ['*'],
    'workflow-admin': lambda workflow: [
        f'workflows:{workflow}:read',
        f'workflows:{workflow}:execute',
        f'templates:{workflow}:write',
        f'presentations:generate',
        f'results:write'
    ],
    'workflow-user': lambda workflow: [
        f'workflows:{workflow}:read',
        f'workflows:{workflow}:execute',
        f'templates:{workflow}:read',
        f'presentations:read',
        f'results:read'
    ],
    'api-only': [
        'presentations:generate',
        'presentations:read',
        'results:read'
    ]
}

def create_user(email: str, name: str, password: str,
                scopes: List[str] = None, preset: str = None,
                workflow: str = None):
    """Create a new user."""
    from app.auth.users.models import User
    from app.util.database import SessionLocal
    import bcrypt

    db = SessionLocal()
    try:
        # Check if user exists
        existing = db.query(User).filter(User.email == email).first()
        if existing:
            print_error(f"User {email} already exists")
            return

        # Determine scopes
        if preset:
            if preset in ['workflow-admin', 'workflow-user']:
                if not workflow:
                    print_error("--workflow required for this preset")
                    return
                user_scopes = SCOPE_PRESETS[preset](workflow)
            else:
                user_scopes = SCOPE_PRESETS[preset]
        else:
            user_scopes = scopes or []

        # Hash password
        password_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()

        # Create user
        user = User(
            email=email,
            name=name,
            password=password_hash,
            scopes=user_scopes,
            active=True
        )

        db.add(user)
        db.commit()

        print_success(f"User {email} created successfully!")
        print_field("Scopes", ', '.join(user_scopes))

    finally:
        db.close()

# In config.py - keep ONE emergency admin for bootstrapping
EMERGENCY_ADMIN_EMAIL = os.environ.get("EMERGENCY_ADMIN_EMAIL")
EMERGENCY_ADMIN_PASSWORD = os.environ.get("EMERGENCY_ADMIN_PASSWORD")

# In auth router
async def login(username: str, password: str):
    # Check emergency admin first (if configured)
    if (settings.EMERGENCY_ADMIN_EMAIL and
        username == settings.EMERGENCY_ADMIN_EMAIL and
        password == settings.EMERGENCY_ADMIN_PASSWORD):

        return SessionData(
            email=username,
            name="Emergency Admin",
            scopes=["*"]  # Full access
        )

    # Otherwise check database
    user = db.query(User).filter(User.email == username).first()
    # ... normal auth flow

# tests/unit/auth/test_scopes.py

def test_scope_checking():
    """Test basic scope checking."""
    auth = BaseAuth()
    auth.scopes = ["templates:read", "presentations:generate"]

    assert auth.has_scope("templates:read") == True
    assert auth.has_scope("templates:write") == False
    assert auth.has_scope("presentations:generate") == True

def test_wildcard_scope():
    """Test wildcard scope grants everything."""
    auth = BaseAuth()
    auth.scopes = ["*"]

    assert auth.has_scope("anything") == True
    assert auth.has_scope("workflows:esg2:execute") == True

def test_workflow_specific_scope():
    """Test workflow-specific access."""
    auth = BaseAuth()
    auth.scopes = ["workflows:esg2:read", "workflows:esg2:execute"]

    assert auth.has_workflow_access("esg2", "read") == True
    assert auth.has_workflow_access("esg2", "execute") == True
    assert auth.has_workflow_access("esg3", "read") == False

def test_resource_access_with_workflow():
    """Test resource access with workflow scoping."""
    auth = BaseAuth()
    auth.scopes = ["templates:esg2:write", "templates:read"]

    # Has workflow-specific write
    assert auth.has_resource_access("templates", "write", "esg2") == True
    # Does not have workflow-specific write for other workflow
    assert auth.has_resource_access("templates", "write", "esg3") == False
    # Has global read
    assert auth.has_resource_access("templates", "read", "esg3") == True

def test_scope_migration():
    """Test converting old attributes to scopes."""
    attributes = {
        "workflows": ["esg2", "presales_poc"],
        "templates": ["view", "admin"],
        "contexts": ["view"]
    }

    scopes = convert_attributes_to_scopes(attributes)

    assert "workflows:esg2:read" in scopes
    assert "workflows:esg2:execute" in scopes
    assert "workflows:presales_poc:read" in scopes
    assert "templates:read" in scopes
    assert "templates:write" in scopes
    assert "contexts:read" in scopes
    assert "contexts:write" not in scopes

# User in config.py
"user@example.com": {
    "id": -100,
    "attributes": {
        "workflows": ["esg2"],
        "templates": ["view", "admin"]
    }
}