from __future__ import annotations import os import time import uuid import logging from typing import Dict, List, Optional import urllib.parse as urlparse import requests from models.schemas import JobPosting, UserProfile, WorkExperience, Education # Set up logging logger = logging.getLogger(__name__) LINKEDIN_AUTH_URL = "https://www.linkedin.com/oauth/v2/authorization" LINKEDIN_TOKEN_URL = "https://www.linkedin.com/oauth/v2/accessToken" class LinkedInClient: def __init__(self) -> None: self.client_id = os.getenv("LINKEDIN_CLIENT_ID", "") self.client_secret = os.getenv("LINKEDIN_CLIENT_SECRET", "") self.redirect_uri = os.getenv("LINKEDIN_REDIRECT_URI", "http://localhost:8501") self.mock_mode = os.getenv("MOCK_MODE", "true").lower() == "true" self.access_token: Optional[str] = None self.state: Optional[str] = None self._stored_state: Optional[str] = None # Store state for validation def get_authorize_url(self) -> str: """Generate LinkedIn OAuth authorization URL with CSRF protection.""" self.state = uuid.uuid4().hex self._stored_state = self.state # Store for validation params = { "response_type": "code", "client_id": self.client_id, "redirect_uri": self.redirect_uri, "state": self.state, "scope": "openid profile email", # Updated to current LinkedIn OAuth 2.0 scopes } auth_url = f"{LINKEDIN_AUTH_URL}?{urlparse.urlencode(params)}" logger.info(f"Generated auth URL with state: {self.state[:8]}...") return auth_url def validate_state(self, state: str) -> bool: """Validate OAuth state parameter to prevent CSRF attacks.""" if not self._stored_state: logger.error("No stored state found for validation") return False if state != self._stored_state: logger.error(f"State mismatch: expected {self._stored_state[:8]}..., got {state[:8]}...") return False logger.info("State validation successful") return True def exchange_code_for_token(self, code: str, state: Optional[str] = None) -> bool: """Exchange authorization code for access token with state validation.""" if self.mock_mode or not (self.client_id and self.client_secret): self.access_token = f"mock_token_{int(time.time())}" logger.info("Using mock mode for authentication") return True # Validate state if provided if state and not self.validate_state(state): logger.error("State validation failed - possible CSRF attack") return False data = { "grant_type": "authorization_code", "code": code, "redirect_uri": self.redirect_uri, "client_id": self.client_id, "client_secret": self.client_secret, } try: logger.info("Exchanging authorization code for access token...") resp = requests.post(LINKEDIN_TOKEN_URL, data=data, timeout=20) if resp.ok: token_data = resp.json() self.access_token = token_data.get("access_token") if self.access_token: logger.info("Successfully obtained access token") # Clear stored state after successful exchange self._stored_state = None return True else: logger.error("No access token in response") return False else: logger.error(f"Token exchange failed: {resp.status_code} - {resp.text}") return False except requests.RequestException as e: logger.error(f"Network error during token exchange: {e}") return False except Exception as e: logger.error(f"Unexpected error during token exchange: {e}") return False def get_profile(self) -> UserProfile: if self.mock_mode or not self.access_token: return UserProfile( full_name="Alex Candidate", headline="Senior Software Engineer", email="alex@example.com", location="Remote", skills=["Python", "AWS", "Docker", "Kubernetes", "PostgreSQL", "Data Engineering"], experiences=[ WorkExperience( title="Senior Software Engineer", company="Acme Inc.", start_date="2021", end_date="Present", achievements=[ "Led migration to AWS, reducing infra costs by 30%", "Implemented CI/CD pipelines with GitHub Actions", ], technologies=["Python", "AWS", "Docker", "Kubernetes"], ), WorkExperience( title="Software Engineer", company="Beta Corp", start_date="2018", end_date="2021", achievements=[ "Built data processing pipelines handling 1B+ events/day", "Optimized Postgres queries cutting latency by 40%", ], technologies=["Python", "PostgreSQL", "Airflow"], ), ], education=[ Education(school="State University", degree="BSc", field_of_study="Computer Science", end_date="2018"), ], links={"GitHub": "https://github.com/example", "LinkedIn": "https://linkedin.com/in/example"}, ) # Minimal profile call (LinkedIn APIs are limited; real app needs compliance) headers = {"Authorization": f"Bearer {self.access_token}"} resp = requests.get("https://api.linkedin.com/v2/me", headers=headers, timeout=20) if not resp.ok: raise RuntimeError("Failed to fetch LinkedIn profile") me = resp.json() # Only basic mapping due to API limitations return UserProfile(full_name=me.get("localizedFirstName", "") + " " + me.get("localizedLastName", "")) def get_saved_jobs(self) -> List[JobPosting]: if self.mock_mode or not self.access_token: return [ JobPosting( id="job_mock_1", title="Senior Data Engineer", company="Nimbus Analytics", location="Remote", description=( "We seek a Senior Data Engineer with Python, AWS (S3, Glue, EMR), Spark, Airflow, and SQL. " "Responsibilities include building scalable data pipelines, CI/CD, and Kubernetes-based deployments." ), url="https://www.linkedin.com/jobs/view/123456", source="mock", saved_by_user=True, ), JobPosting( id="job_mock_2", title="Platform Engineer", company="Orion Cloud", location="London, UK", description=( "Looking for a Platform Engineer skilled in Docker, Kubernetes, Terraform, AWS, observability (Prometheus, Grafana), and security best practices." ), url="https://www.linkedin.com/jobs/view/654321", source="mock", saved_by_user=True, ), ] # Placeholder: LinkedIn Jobs API access is restricted; this is a stub return [] def get_job_details(self, job_id: str) -> Optional[JobPosting]: jobs = self.get_saved_jobs() for job in jobs: if job.id == job_id: return job return None