"""Views for GitHub OAuth callbacks and webhooks.""" import logging from typing import Any import requests from django.conf import settings from django.http import HttpRequest, HttpResponse, HttpResponseRedirect, JsonResponse from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_GET, require_POST from accounts.models import Workspace from integrations.github.services.client import GitHubClient from integrations.models.github import GitHubInstallation logger = logging.getLogger(__name__) @require_GET def github_oauth_callback(request: HttpRequest) -> HttpResponse: """Handle GitHub OAuth callback after user authorizes the GitHub App. Expected query parameters: - code: The authorization code from GitHub (for OAuth flow) - state: The state parameter containing signed workspace_id - installation_id: The GitHub App installation ID - setup_action: The action taken (install, update, request) This endpoint handles the redirect from GitHub after a user installs or configures the GitHub App for their account/organization. """ from django.core.signing import BadSignature, SignatureExpired, TimestampSigner installation_id = request.GET.get("installation_id") setup_action = request.GET.get("setup_action", "install") state = request.GET.get("state") # Get the frontend URL for redirects frontend_url = settings.CORS_ALLOWED_ORIGINS[0] if settings.CORS_ALLOWED_ORIGINS else "http://localhost:3000" if not installation_id: logger.warning("GitHub OAuth callback missing installation_id") return HttpResponseRedirect(f"{frontend_url}/dashboard/sources?error=missing_installation_id") # Validate state and extract workspace_id # State format: {random_token}:{signed_workspace_id} # The signed_workspace_id is a Django TimestampSigner token workspace = None if state and ":" in state: try: _, signed_workspace_id = state.split(":", 1) signer = TimestampSigner() # Allow state to be valid for 1 hour (3600 seconds) workspace_id = signer.unsign(signed_workspace_id, max_age=3600) workspace = Workspace.objects.get(pk=workspace_id) except (ValueError, BadSignature, SignatureExpired) as e: logger.warning(f"GitHub OAuth callback: invalid state signature - {e}") except Workspace.DoesNotExist: logger.warning(f"GitHub OAuth callback: workspace not found - {workspace_id}") if not workspace: logger.warning("GitHub OAuth callback: could not validate state") return HttpResponseRedirect(f"{frontend_url}/dashboard/sources?error=invalid_state") if setup_action == "request": # User requested access but doesn't have permission to install return HttpResponseRedirect(f"{frontend_url}/dashboard/sources?error=installation_requested") try: # Fetch installation details from GitHub API installation_data = _fetch_installation_details(int(installation_id)) # Create or update the installation record installation, created = GitHubInstallation.objects.update_or_create( workspace=workspace, defaults={ "installation_id": int(installation_id), "account_login": installation_data["account"]["login"], "account_type": installation_data["account"]["type"], "account_id": installation_data["account"]["id"], "account_avatar_url": installation_data["account"].get("avatar_url"), "permissions": installation_data.get("permissions", {}), }, ) # Refresh the access token for the new installation client = GitHubClient(installation) client._refresh_installation_token() action = "connected" if created else "updated" logger.info(f"GitHub installation {action}: {installation.account_login} " f"(workspace: {workspace.slug})") return HttpResponseRedirect(f"{frontend_url}/dashboard/sources?github={action}") except Exception as e: logger.exception(f"Error processing GitHub OAuth callback: {e}") return HttpResponseRedirect(f"{frontend_url}/dashboard/sources?error=installation_failed") def _fetch_installation_details(installation_id: int) -> dict[str, Any]: """Fetch installation details from GitHub API using JWT auth. Args: installation_id: The GitHub installation ID Returns: The installation data from GitHub API """ import base64 import time import jwt app_id = settings.GITHUB_APP_ID private_key_base64 = settings.GITHUB_APP_PRIVATE_KEY_BASE64 if not app_id or not private_key_base64: raise ValueError("GitHub App credentials not configured") # Decode private key private_key = base64.b64decode(private_key_base64).decode("utf-8") # Generate JWT now = int(time.time()) payload = { "iat": now - 60, "exp": now + (10 * 60), "iss": app_id, } jwt_token = jwt.encode(payload, private_key, algorithm="RS256") # Fetch installation details response = requests.get( f"https://api.github.com/app/installations/{installation_id}", headers={ "Authorization": f"Bearer {jwt_token}", "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28", }, timeout=30, ) if response.status_code != 200: raise ValueError(f"Failed to fetch installation details: {response.status_code} {response.text}") result: dict[str, Any] = response.json() return result @csrf_exempt @require_POST def github_webhook(request: HttpRequest) -> HttpResponse: """Handle GitHub webhook events. Validates webhook signature and processes events for: - issues (opened, closed, reopened, labeled, commented) - pull_request (opened, closed, merged, reviewed) - discussion (created, answered, commented) - installation (created, deleted, suspended) """ import hashlib import hmac # Verify webhook signature signature_header = request.headers.get("X-Hub-Signature-256", "") if not signature_header: logger.warning("GitHub webhook missing signature") return JsonResponse({"error": "Missing signature"}, status=401) webhook_secret = settings.GITHUB_WEBHOOK_SECRET if not webhook_secret: logger.error("GITHUB_WEBHOOK_SECRET not configured") return JsonResponse({"error": "Webhook not configured"}, status=500) # Compute expected signature expected_signature = ( "sha256=" + hmac.new( webhook_secret.encode("utf-8"), request.body, hashlib.sha256, ).hexdigest() ) if not hmac.compare_digest(signature_header, expected_signature): logger.warning("GitHub webhook signature mismatch") return JsonResponse({"error": "Invalid signature"}, status=401) # Parse event event_type = request.headers.get("X-GitHub-Event", "") delivery_id = request.headers.get("X-GitHub-Delivery", "") try: import json payload = json.loads(request.body) except json.JSONDecodeError: return JsonResponse({"error": "Invalid JSON"}, status=400) logger.info(f"GitHub webhook received: {event_type} (delivery: {delivery_id})") # Handle all events through the webhook handler handled_events = ( "installation", "installation_repositories", "issues", "pull_request", "discussion", "issue_comment", "pull_request_review", "pull_request_review_comment", ) if event_type in handled_events: return _handle_activity_event(event_type, payload) # Unknown event type - acknowledge but don't process return JsonResponse({"status": "ignored", "event": event_type}) def _handle_installation_event(payload: dict) -> JsonResponse: """Handle GitHub App installation events.""" action = payload.get("action") installation_data = payload.get("installation", {}) installation_id = installation_data.get("id") if not installation_id: return JsonResponse({"error": "Missing installation ID"}, status=400) try: installation = GitHubInstallation.objects.get(installation_id=installation_id) except GitHubInstallation.DoesNotExist: # Installation not in our database - might be for a different app instance logger.info(f"Webhook for unknown installation: {installation_id}") return JsonResponse({"status": "ignored", "reason": "unknown_installation"}) if action == "deleted": # Installation was removed logger.info(f"GitHub installation deleted: {installation.account_login}") installation.repositories.update(is_archived=True, sync_status="disabled") installation.delete() elif action == "suspend": # Installation was suspended from django.utils import timezone logger.info(f"GitHub installation suspended: {installation.account_login}") installation.suspended_at = timezone.now() installation.repositories.update(sync_status="disabled") installation.save(update_fields=["suspended_at", "updated_at"]) elif action == "unsuspend": # Installation was unsuspended logger.info(f"GitHub installation unsuspended: {installation.account_login}") installation.suspended_at = None installation.repositories.filter(is_archived=False).update(sync_status="active") installation.save(update_fields=["suspended_at", "updated_at"]) elif action in ("created", "new_permissions_accepted"): # Update permissions installation.permissions = installation_data.get("permissions", {}) installation.save(update_fields=["permissions", "updated_at"]) return JsonResponse({"status": "processed", "action": action}) def _handle_activity_event(event_type: str, payload: dict) -> JsonResponse: """Handle GitHub activity events (issues, PRs, discussions).""" from integrations.webhooks import process_webhook result = process_webhook(event_type, payload) return JsonResponse(result)