Spaces:
Runtime error
Runtime error
File size: 8,310 Bytes
7498f2c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 |
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 |