Merge pull request #90 from MyElectricalData/develop

Develop

Author: m4dm4rtig4n

CHANGELOG.md

@@ -1,5 +1,21 @@
 # Changelog
 
+## [1.4.0-dev.2](https://github.com/MyElectricalData/myelectricaldata_new/compare/1.4.0-dev.1...1.4.0-dev.2) (2025-12-20)
+
+### Bug Fixes
+
+* address Copilot review suggestions ([f78115a](https://github.com/MyElectricalData/myelectricaldata_new/commit/f78115a8dbed44f5a03fd410f8be5da01b2554db))
+
+## [1.4.0-dev.1](https://github.com/MyElectricalData/myelectricaldata_new/compare/1.3.0...1.4.0-dev.1) (2025-12-20)
+
+### Features
+
+* add admin data sharing for PDL debugging ([cfb32b8](https://github.com/MyElectricalData/myelectricaldata_new/commit/cfb32b83ddb435c9c792aec031ead2ad2ae054ab))
+
+### Bug Fixes
+
+* **api:** correct type annotation for cached_data in admin router ([9b383f1](https://github.com/MyElectricalData/myelectricaldata_new/commit/9b383f13dade4ea6332254165e075837c17fd26b))
+
 ## [1.3.0](https://github.com/MyElectricalData/myelectricaldata_new/compare/1.2.0...1.3.0) (2025-12-20)
 
 ### Features

apps/api/alembic/versions/20251220_0100_007_add_admin_data_sharing.py

@@ -0,0 +1,30 @@
+"""add admin_data_sharing to users
+
+Revision ID: 007
+Revises: 006
+Create Date: 2025-12-20
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '007'
+down_revision: Union[str, None] = '006'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Add admin_data_sharing columns to users table
+    op.add_column('users', sa.Column('admin_data_sharing', sa.Boolean(), nullable=False, server_default='false'))
+    op.add_column('users', sa.Column('admin_data_sharing_enabled_at', sa.DateTime(timezone=True), nullable=True))
+
+
+def downgrade() -> None:
+    # Remove admin_data_sharing columns from users table
+    op.drop_column('users', 'admin_data_sharing_enabled_at')
+    op.drop_column('users', 'admin_data_sharing')

apps/api/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "myelectricaldata-api"
-version = "1.3.0"
+version = "1.4.0-dev.2"
 description = "MyElectricalData API Gateway for Enedis data"
 authors = [{name = "m4dm4rtig4n"}]
 license = {text = "Apache-2.0"}

apps/api/src/middleware/__init__.py

@@ -1,4 +1,11 @@
-from .auth import get_current_user, require_not_demo, is_demo_user, DEMO_EMAIL
+from .auth import (
+    get_current_user,
+    require_not_demo,
+    is_demo_user,
+    DEMO_EMAIL,
+    get_impersonation_context,
+    get_encryption_key,
+)
 from .admin import require_admin, require_permission, require_action
 
 __all__ = [
@@ -9,4 +16,6 @@
     "require_not_demo",
     "is_demo_user",
     "DEMO_EMAIL",
+    "get_impersonation_context",
+    "get_encryption_key",
 ]

apps/api/src/middleware/auth.py

@@ -158,3 +158,61 @@ async def require_not_demo(current_user: User = Depends(get_current_user)) -> Us
             detail="Le compte de démonstration est en lecture seule"
         )
     return current_user
+
+
+async def get_impersonation_context(
+    request: Request,
+    current_user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+) -> Optional[User]:
+    """
+    Get the user being impersonated if admin is accessing shared data.
+
+    Returns:
+        - The impersonated user (for getting their client_secret) if:
+          1. Current user is admin (is_admin or role admin)
+          2. X-Impersonate-User-Id header is present
+          3. Target user has enabled admin_data_sharing
+        - None otherwise (use current_user's client_secret)
+
+    This is used to properly decrypt cached data that was encrypted
+    with the data owner's client_secret.
+    """
+    impersonate_user_id = request.headers.get("X-Impersonate-User-Id")
+
+    if not impersonate_user_id:
+        return None
+
+    # Check if current user is admin
+    is_admin = current_user.is_admin or (current_user.role and current_user.role.name == "admin")
+    if not is_admin:
+        logger.warning(f"[IMPERSONATION] Non-admin user {current_user.email} tried to impersonate {impersonate_user_id}")
+        return None
+
+    # Get the target user
+    result = await db.execute(select(User).where(User.id == impersonate_user_id))
+    target_user = result.scalar_one_or_none()
+
+    if not target_user:
+        logger.warning(f"[IMPERSONATION] Admin {current_user.email} tried to impersonate non-existent user {impersonate_user_id}")
+        return None
+
+    # Check if target user has enabled data sharing
+    if not target_user.admin_data_sharing:
+        logger.warning(f"[IMPERSONATION] Admin {current_user.email} tried to impersonate user {target_user.email} who has not enabled data sharing")
+        return None
+
+    logger.info(f"[IMPERSONATION] Admin {current_user.email} impersonating user {target_user.email}")
+    return target_user
+
+
+def get_encryption_key(current_user: User, impersonated_user: Optional[User]) -> str:
+    """
+    Get the encryption key (client_secret) to use for cache operations.
+
+    Uses the impersonated user's key if impersonation is active,
+    otherwise uses the current user's key.
+    """
+    if impersonated_user:
+        return impersonated_user.client_secret
+    return current_user.client_secret

apps/api/src/models/user.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 import uuid
+from datetime import datetime
 from typing import TYPE_CHECKING
-from sqlalchemy import String, Boolean, ForeignKey
+from sqlalchemy import String, Boolean, ForeignKey, DateTime
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 from .base import Base, TimestampMixin
 
@@ -23,6 +24,8 @@ class User(Base, TimestampMixin):
     is_admin: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)  # Kept for backward compatibility
     email_verified: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
     debug_mode: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
+    admin_data_sharing: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
+    admin_data_sharing_enabled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     enedis_customer_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
     role_id: Mapped[str | None] = mapped_column(String(36), ForeignKey("roles.id"), nullable=True)
 

apps/api/src/routers/accounts.py

@@ -276,11 +276,13 @@ async def get_current_user_info(
         created_at=user.created_at,
     )
 
-    # Add is_admin field, debug_mode, and role
+    # Add is_admin field, debug_mode, admin_data_sharing, and role
     # is_admin is true if: database flag OR email in ADMIN_EMAILS env var
     user_data = user_response.model_dump()
     user_data['is_admin'] = user.is_admin or settings.is_admin(user.email)
     user_data['debug_mode'] = user.debug_mode
+    user_data['admin_data_sharing'] = user.admin_data_sharing
+    user_data['admin_data_sharing_enabled_at'] = user.admin_data_sharing_enabled_at.isoformat() if user.admin_data_sharing_enabled_at else None
 
     # Add role information with permissions
     if user.role:
@@ -592,6 +594,38 @@ async def reset_password(request: Request, db: AsyncSession = Depends(get_db)) -
     return APIResponse(success=True, data={"message": "Password reset successfully!"})
 
 
+@router.post("/toggle-admin-sharing", response_model=APIResponse)
+async def toggle_admin_data_sharing(
+    current_user: User = Depends(require_not_demo),
+    db: AsyncSession = Depends(get_db)
+) -> APIResponse:
+    """Toggle admin data sharing for current user.
+
+    When enabled, administrators can access the user's PDL data and cache
+    for debugging purposes. The user can revoke this access at any time.
+    """
+    current_user.admin_data_sharing = not current_user.admin_data_sharing
+
+    if current_user.admin_data_sharing:
+        current_user.admin_data_sharing_enabled_at = datetime.now(UTC)
+    else:
+        current_user.admin_data_sharing_enabled_at = None
+
+    await db.commit()
+
+    action = "enabled" if current_user.admin_data_sharing else "disabled"
+    logger.info(f"[ADMIN_SHARING] User {current_user.email} {action} admin data sharing")
+
+    return APIResponse(
+        success=True,
+        data={
+            "admin_data_sharing": current_user.admin_data_sharing,
+            "admin_data_sharing_enabled_at": current_user.admin_data_sharing_enabled_at.isoformat() if current_user.admin_data_sharing_enabled_at else None,
+            "message": f"Partage des données avec les administrateurs {'activé' if current_user.admin_data_sharing else 'désactivé'}"
+        }
+    )
+
+
 @router.post("/update-password", response_model=APIResponse)
 async def update_password(
     request: Request,