ACR-30 Implementation of auth module

.github/workflows/reviewer.yml

@@ -1,43 +1,26 @@
-name: PR Reviewer Agent
+name: PR Event Logger
+
 on:
-  issue_comment:
-    types: [created]
   pull_request:
-    types: [opened, synchronize, reopened]
-  push:
+    types: [opened, reopened, ready_for_review, review_requested]
+  issue_comment:
+    types: [created, edited]
+
 jobs:
-  process_pr_events:
+  log-event:
     runs-on: ubuntu-latest
     steps:
-      - name: Extract event details
-        run: echo "EVENT_PAYLOAD=$(jq -c . < $GITHUB_EVENT_PATH)" >> $GITHUB_ENV
-
-      - name: Generate Signature and Encrypt Token
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: '3.9'
+
+      - name: Run event logger
         env:
-          WEBHOOK_SECRET: ${{ secrets.WEBHOOK_SECRET }}
-          API_TOKEN: ${{ secrets.API_TOKEN }}
-        run: |
-          # Generate signature for the payload
-          SIGNATURE=$(echo -n "$EVENT_PAYLOAD" | openssl dgst -sha256 -hmac "$WEBHOOK_SECRET" | cut -d " " -f2)
-          echo "SIGNATURE=$SIGNATURE" >> $GITHUB_ENV
-          
-          # Create a consistent key from the webhook secret
-          KEY=$(echo -n "$WEBHOOK_SECRET" | openssl dgst -sha256 | cut -d ' ' -f2)
-          
-          # Generate a random IV
-          IV=$(openssl rand -hex 16)
-          
-          # Encrypt token with proper padding
-          ENCRYPTED_TOKEN=$(echo -n "$API_TOKEN" | openssl enc -aes-256-cbc -a -A -K "$KEY" -iv "$IV" -md sha256)
-          
-          echo "ENCRYPTED_TOKEN=$ENCRYPTED_TOKEN" >> $GITHUB_ENV
-          echo "TOKEN_IV=$IV" >> $GITHUB_ENV
-          
-      - name: Call External API (With Encrypted Token)
-        run: |
-          curl -X POST https://firstly-worthy-chamois.ngrok-free.app/github-webhook \
-          -H "Content-Type: application/json" \
-          -H "X-Hub-Signature-256: sha256=$SIGNATURE" \
-          -H "X-Encrypted-Token: $ENCRYPTED_TOKEN" \
-          -H "X-Token-IV: $TOKEN_IV" \
-          -d "$EVENT_PAYLOAD"
+          GITHUB_EVENT_NAME: ${{ github.event_name }}
+          GITHUB_EVENT_PATH: ${{ github.event_path }}
+        run: python main.py
+

app/__init__.py

@@ -0,0 +1 @@
+# Authentication API package 

app/auth/__init__.py

@@ -0,0 +1 @@
+# Auth module 

app/auth/dependencies.py

@@ -0,0 +1,67 @@
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from typing import Optional, Dict, Any
+from .firebase_auth import firebase_auth
+
+# Security scheme for Bearer token
+security = HTTPBearer()
+
+
+async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)) -> Dict[str, Any]:
+    """
+    Dependency to get current authenticated user from Firebase token
+    """
+    token = credentials.credentials
+
+    if not token:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Could not validate credentials",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    user_data = await firebase_auth.verify_token(token)
+
+    if not user_data:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid authentication credentials",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    return user_data
+
+
+async def get_current_active_user(current_user: Dict[str, Any] = Depends(get_current_user)) -> Dict[str, Any]:
+    """
+    Dependency to get current active user
+    """
+    if not current_user.get("is_active", True):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Inactive user"
+        )
+    return current_user
+
+
+async def require_role(required_role: str):
+    """
+    Dependency factory to require specific role
+    """
+    async def role_checker(current_user: Dict[str, Any] = Depends(get_current_user)) -> Dict[str, Any]:
+        user_role = current_user.get("role", "user")
+
+        if user_role != required_role and user_role != "admin":
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail=f"Access denied. Required role: {required_role}"
+            )
+
+        return current_user
+
+    return role_checker
+
+
+# Predefined role dependencies
+require_admin = require_role("admin")
+require_user = require_role("user") 

app/auth/firebase_auth.py

@@ -0,0 +1,158 @@
+import os
+import firebase_admin
+from firebase_admin import auth, credentials
+from firebase_admin.auth import UserRecord
+from typing import Optional, Dict, Any
+import json
+from datetime import datetime, timedelta
+import jwt
+
+
+class FirebaseAuthService:
+    def __init__(self):
+        self._initialize_firebase()
+        self.jwt_secret = os.getenv("JWT_SECRET", "your-secret-key")
+        self.jwt_algorithm = "HS256"
+        self.access_token_expiry = timedelta(hours=1)
+        self.refresh_token_expiry = timedelta(days=7)
+
+    def _initialize_firebase(self):
+        """Initialize Firebase Admin SDK"""
+        try:
+            # Try to get Firebase credentials from environment
+            firebase_credentials = os.getenv("FIREBASE_CREDENTIALS")
+            if firebase_credentials:
+                cred_dict = json.loads(firebase_credentials)
+                cred = credentials.Certificate(cred_dict)
+            else:
+                # Fallback to service account file
+                service_account_path = os.getenv("FIREBASE_SERVICE_ACCOUNT_PATH")
+                if service_account_path and os.path.exists(service_account_path):
+                    cred = credentials.Certificate(service_account_path)
+                else:
+                    # Use default credentials (for development)
+                    cred = credentials.ApplicationDefault()
+
+            firebase_admin.initialize_app(cred)
+        except Exception as e:
+            print(f"Firebase initialization error: {e}")
+            raise
+
+    async def create_user(self, email: str, password: str, first_name: str, last_name: str) -> Dict[str, Any]:
+        """Create a new user in Firebase"""
+        try:
+            user_record = auth.create_user(
+                email=email,
+                password=password,
+                display_name=f"{first_name} {last_name}",
+                email_verified=False
+            )
+
+            # Set custom claims
+            auth.set_custom_user_claims(user_record.uid, {
+                "first_name": first_name,
+                "last_name": last_name,
+                "role": "user"
+            })
+
+            return {
+                "id": user_record.uid,
+                "email": user_record.email,
+                "first_name": first_name,
+                "last_name": last_name,
+                "is_active": not user_record.disabled,
+                "created_at": user_record.user_metadata.creation_timestamp
+            }
+        except Exception as e:
+            raise Exception(f"Failed to create user: {str(e)}")
+
+    async def sign_in_user(self, email: str, password: str) -> Dict[str, Any]:
+        """Sign in user with email and password"""
+        try:
+            # In a real implementation, you would use Firebase Auth REST API
+            # For now, we'll simulate the authentication
+            user_record = auth.get_user_by_email(email)
+
+            if user_record.disabled:
+                raise Exception("User account is disabled")
+
+            # Generate JWT tokens
+            access_token = self._generate_access_token(user_record.uid, user_record.email)
+            refresh_token = self._generate_refresh_token(user_record.uid)
+
+            # Get custom claims
+            custom_claims = auth.get_custom_user_claims(user_record.uid)
+
+            return {
+                "access_token": access_token,
+                "refresh_token": refresh_token,
+                "user": {
+                    "id": user_record.uid,
+                    "email": user_record.email,
+                    "first_name": custom_claims.get("first_name", ""),
+                    "last_name": custom_claims.get("last_name", ""),
+                    "is_active": not user_record.disabled,
+                    "created_at": str(user_record.user_metadata.creation_timestamp)
+                }
+            }
+        except Exception as e:
+            raise Exception(f"Authentication failed: {str(e)}")
+
+    async def verify_token(self, token: str) -> Optional[Dict[str, Any]]:
+        """Verify Firebase ID token"""
+        try:
+            decoded_token = auth.verify_id_token(token)
+            user_record = auth.get_user(decoded_token["uid"])
+            custom_claims = auth.get_custom_user_claims(user_record.uid)
+
+            return {
+                "uid": user_record.uid,
+                "email": user_record.email,
+                "first_name": custom_claims.get("first_name", ""),
+                "last_name": custom_claims.get("last_name", ""),
+                "role": custom_claims.get("role", "user")
+            }
+        except Exception as e:
+            print(f"Token verification failed: {e}")
+            return None
+
+    def _generate_access_token(self, user_id: str, email: str) -> str:
+        """Generate JWT access token"""
+        payload = {
+            "user_id": user_id,
+            "email": email,
+            "exp": datetime.utcnow() + self.access_token_expiry,
+            "iat": datetime.utcnow(),
+            "type": "access"
+        }
+        return jwt.encode(payload, self.jwt_secret, algorithm=self.jwt_algorithm)
+
+    def _generate_refresh_token(self, user_id: str) -> str:
+        """Generate JWT refresh token"""
+        payload = {
+            "user_id": user_id,
+            "exp": datetime.utcnow() + self.refresh_token_expiry,
+            "iat": datetime.utcnow(),
+            "type": "refresh"
+        }
+        return jwt.encode(payload, self.jwt_secret, algorithm=self.jwt_algorithm)
+
+    async def refresh_access_token(self, refresh_token: str) -> Optional[str]:
+        """Refresh access token using refresh token"""
+        try:
+            payload = jwt.decode(refresh_token, self.jwt_secret, algorithms=[self.jwt_algorithm])
+
+            if payload.get("type") != "refresh":
+                raise Exception("Invalid token type")
+
+            user_id = payload.get("user_id")
+            user_record = auth.get_user(user_id)
+
+            return self._generate_access_token(user_id, user_record.email)
+        except Exception as e:
+            print(f"Token refresh failed: {e}")
+            return None
+
+
+# Global instance
+firebase_auth = FirebaseAuthService() 

app/auth/models.py

@@ -0,0 +1,39 @@
+from pydantic import BaseModel, EmailStr
+from typing import Optional
+
+
+class UserSignupRequest(BaseModel):
+    email: EmailStr
+    password: str
+    first_name: str
+    last_name: str
+
+
+class UserLoginRequest(BaseModel):
+    email: EmailStr
+    password: str
+
+
+class UserResponse(BaseModel):
+    id: str
+    email: str
+    first_name: str
+    last_name: str
+    is_active: bool
+    created_at: str
+
+
+class AuthResponse(BaseModel):
+    access_token: str
+    refresh_token: str
+    token_type: str = "bearer"
+    user: UserResponse
+
+
+class TokenResponse(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+
+
+class RefreshTokenRequest(BaseModel):
+    refresh_token: str