monica_site/site/social/oauth_linkedin.py

"""LinkedIn 3-legged OAuth helpers (Authorization Code flow)."""

from __future__ import annotations

import logging
from datetime import timedelta
from urllib.parse import urlencode

import requests
from django.conf import settings
from django.urls import reverse
from django.utils import timezone

from social.crypto import decrypt_tokens
from social.models import Platform, SocialAppCredentials

logger = logging.getLogger(__name__)

AUTHORIZE_URL = "https://www.linkedin.com/oauth/v2/authorization"
TOKEN_URL = "https://www.linkedin.com/oauth/v2/accessToken"
USERINFO_URL = "https://api.linkedin.com/v2/userinfo"
ME_URL = "https://api.linkedin.com/v2/me"

# openid/profile/email: Sign In with LinkedIn (OpenID Connect)
# w_member_social: Share on LinkedIn product (required to post as the member)
DEFAULT_SCOPES = ("openid", "profile", "email", "w_member_social")


class LinkedInOAuthError(RuntimeError):
    """Raised when LinkedIn OAuth or profile lookup fails."""


def get_app_credentials() -> SocialAppCredentials | None:
    return SocialAppCredentials.objects.filter(platform=Platform.LINKEDIN).first()


def load_client_credentials() -> tuple[str, str]:
    """Return (client_id, client_secret) from portal-saved credentials."""
    app = get_app_credentials()
    if not app or not app.client_id or not app.encrypted_client_secret:
        raise LinkedInOAuthError(
            "LinkedIn app credentials are not saved yet. "
            "Enter Client ID and Client Secret on the Connect LinkedIn form."
        )
    try:
        secret = decrypt_tokens(app.encrypted_client_secret)
    except ValueError as exc:
        raise LinkedInOAuthError(
            "Could not decrypt LinkedIn client secret. "
            "Re-enter the Client Secret on the Connect LinkedIn form."
        ) from exc
    if not secret:
        raise LinkedInOAuthError("LinkedIn client secret is empty — save it again.")
    return app.client_id.strip(), secret.strip()


def credentials_configured() -> bool:
    app = get_app_credentials()
    return bool(app and app.is_configured)


def redirect_uri() -> str:
    """Absolute callback URL — must match Authorized redirect URLs in LinkedIn Auth tab."""
    base = (settings.PUBLIC_SITE_URL or "").rstrip("/")
    if not base:
        raise LinkedInOAuthError(
            "PUBLIC_SITE_URL is not set; cannot build LinkedIn OAuth redirect URI."
        )
    return f"{base}{reverse('social:linkedin_oauth_callback')}"


def authorization_url(*, state: str, scopes: tuple[str, ...] = DEFAULT_SCOPES) -> str:
    client_id, _secret = load_client_credentials()
    params = {
        "response_type": "code",
        "client_id": client_id,
        "redirect_uri": redirect_uri(),
        "state": state,
        "scope": " ".join(scopes),
    }
    return f"{AUTHORIZE_URL}?{urlencode(params)}"


def exchange_code(code: str) -> dict:
    """Exchange authorization code for access token payload."""
    client_id, client_secret = load_client_credentials()
    response = requests.post(
        TOKEN_URL,
        data={
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": redirect_uri(),
            "client_id": client_id,
            "client_secret": client_secret,
        },
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        timeout=30,
    )
    if response.status_code >= 400:
        logger.warning(
            "LinkedIn token exchange failed: %s %s",
            response.status_code,
            response.text[:500],
        )
        raise LinkedInOAuthError(
            f"LinkedIn token exchange failed ({response.status_code}). "
            "Check client ID/secret, redirect URI, and that Share on LinkedIn is enabled."
        )
    payload = response.json()
    if not payload.get("access_token"):
        raise LinkedInOAuthError("LinkedIn token response missing access_token.")
    return payload


def fetch_member_profile(access_token: str) -> dict:
    """
    Resolve member display name + person URN.

    Prefers OpenID userinfo (`sub`); falls back to legacy /v2/me.
    """
    headers = {"Authorization": f"Bearer {access_token}"}
    userinfo = requests.get(USERINFO_URL, headers=headers, timeout=30)
    if userinfo.status_code == 200:
        data = userinfo.json()
        sub = (data.get("sub") or "").strip()
        if sub:
            name = (
                (data.get("name") or "").strip()
                or " ".join(
                    p
                    for p in (
                        (data.get("given_name") or "").strip(),
                        (data.get("family_name") or "").strip(),
                    )
                    if p
                )
                or "LinkedIn member"
            )
            return {
                "author_urn": f"urn:li:person:{sub}",
                "label": name,
                "email": (data.get("email") or "").strip(),
            }

    me = requests.get(
        ME_URL,
        headers={
            **headers,
            "X-Restli-Protocol-Version": "2.0.0",
        },
        timeout=30,
    )
    if me.status_code >= 400:
        logger.warning(
            "LinkedIn profile lookup failed: userinfo=%s me=%s %s",
            userinfo.status_code,
            me.status_code,
            me.text[:500],
        )
        raise LinkedInOAuthError(
            "Could not load LinkedIn profile. Ensure Sign In with LinkedIn "
            "(openid/profile) is approved on the app."
        )
    data = me.json()
    person_id = (data.get("id") or "").strip()
    if not person_id:
        raise LinkedInOAuthError("LinkedIn /v2/me response missing id.")
    localized = data.get("localizedFirstName") or ""
    last = data.get("localizedLastName") or ""
    label = f"{localized} {last}".strip() or "LinkedIn member"
    return {"author_urn": f"urn:li:person:{person_id}", "label": label, "email": ""}


def token_blob_from_oauth(token_payload: dict, *, author_urn: str) -> dict:
    expires_in = int(token_payload.get("expires_in") or 0)
    blob: dict = {
        "access_token": token_payload["access_token"],
        "author_urn": author_urn,
        "token_type": token_payload.get("token_type") or "Bearer",
        "scope": token_payload.get("scope") or "",
    }
    if token_payload.get("refresh_token"):
        blob["refresh_token"] = token_payload["refresh_token"]
    if expires_in:
        blob["expires_at"] = (
            timezone.now() + timedelta(seconds=expires_in)
        ).isoformat()
    return blob