"""Source-tracking tokens for public job application URLs. Public application links used to carry the platform in plain text (`/careers/jobs/?source=linkedin`), which leaked the channel to anyone who saw the link and let a visitor rewrite the value before applying. Every link a provider now generates carries a single opaque token instead: /careers/jobs/?t= Two token formats resolve here. Both hide the platform completely; they differ only in where the payload lives. 1. SHORT CODE (current format, ~10 chars) — a random, unguessable reference to a `JobTrackingLink` row that holds the JD, channel and tracking metadata. The code itself carries no payload, so there is nothing to decode and nothing to tamper with; only the backend can map it back. This is what `build_application_url()` mints today. 2. FERNET TOKEN (legacy, ~180 chars) — the JD/channel/tracking payload encrypted with the project's standard Fernet approach (core.crypto, apps.llm.crypto, apps.candidates.unsubscribe_tokens), keyed separately so a tracking token can never be swapped in for an encrypted-id or unsubscribe token. Kept because links minted in this format are already published on LinkedIn, stored on JobPosting.external_url and sent over WhatsApp/SMS — they must keep working. Fernet's 57-byte envelope is why it can't simply be shortened, hence format 1. Either way the backend resolves the JD reference, the platform channel and the tracking metadata server-side at apply time. An invalid or missing token is never an error — the application is simply treated as Direct, matching the existing business rules. """ import base64 import hashlib import json import logging from django.conf import settings from cryptography.fernet import Fernet, InvalidToken logger = logging.getLogger(__name__) #: Current payload version — lets a future format change stay readable. _VERSION = 1 def _fernet() -> Fernet: """Fernet keyed like core.crypto, but with its own derived key.""" key = getattr(settings, "JOB_TRACKING_ENCRYPTION_KEY", "") or "" if key: return Fernet(key.encode() if isinstance(key, str) else key) digest = hashlib.sha256(f"{settings.SECRET_KEY}::job-tracking".encode()).digest() return Fernet(base64.urlsafe_b64encode(digest)) # --------------------------------------------------------------------------- # Channel -> candidate source # --------------------------------------------------------------------------- #: JobPosting.Channel -> Candidate.SourceChoices. Any channel missing here (a #: newly added job board that has no matching candidate source yet) resolves to #: DIRECT rather than failing, so adding a platform never breaks the apply flow. CHANNEL_SOURCE_MAP = { "LINKEDIN": "LINKEDIN", "NAUKRI": "NAUKRI", "WHATSAPP": "WHATSAPP", "SMS": "SMS", "TELEGRAM": "TELEGRAM", "EMAIL": "EMAIL", "REFERRAL": "REFERRAL", # The career portal *is* the direct channel — the pre-token career-portal # link already carried `?source=direct`, so this preserves that mapping. "CAREER_PORTAL": "DIRECT", "OTHER": "OTHER", } #: Fallback when a token is missing, invalid, or maps to an unknown channel. DEFAULT_SOURCE = "DIRECT" def channel_to_source(channel: str) -> str: """Candidate.SourceChoices value for a JobPosting channel (DIRECT if unknown).""" return CHANNEL_SOURCE_MAP.get((channel or "").upper(), DEFAULT_SOURCE) # --------------------------------------------------------------------------- # Short codes (current format) # --------------------------------------------------------------------------- #: A Fernet token always starts with this (version byte 0x80 in urlsafe-b64). #: Used only to route a token to the right resolver first — both are still #: tried, so the prefix is a hint, never a trust decision. _FERNET_PREFIX = "gAAAAA" def make_short_code(job, channel: str, tracking: dict | None = None, label: str = "", user=None) -> str: """Stable short code for (job, channel, label), creating the row if needed. Returns "" if the code can't be persisted (e.g. an unsaved job), letting the caller fall back to the self-contained Fernet token instead of failing. """ from django.db import IntegrityError, transaction from .models import JobTrackingLink if not getattr(job, "pk", None): return "" for _attempt in range(3): try: with transaction.atomic(): link, created = JobTrackingLink.objects.get_or_create( job=job, channel=(channel or "").upper(), label=label or "", defaults={ "code": JobTrackingLink.generate_code(), "tracking": tracking or {}, "created_by": user, }, ) # Metadata supplied later (a campaign added to an existing link) is # recorded without ever changing the code the recruiter already has. if not created and tracking and link.tracking != tracking: link.tracking = tracking link.save(update_fields=["tracking", "updated_at"]) return link.code except IntegrityError: continue # code collision — regenerate and retry except Exception: # noqa: BLE001 — link generation must never break publishing logger.exception("Could not mint a short tracking code for JD %s / %s", getattr(job, "id", "?"), channel) return "" return "" def resolve_short_code(code: str) -> dict | None: """Look up a short code. Returns the same shape as resolve_tracking_token().""" from .models import JobTrackingLink code = (code or "").strip() if not code or len(code) > 32: return None try: link = JobTrackingLink.objects.select_related("job").filter(code=code).first() except Exception: # noqa: BLE001 — a DB hiccup must not break the apply flow logger.exception("Short tracking code lookup failed") return None if not link: return None job_ref = str(getattr(link.job, "public_id", None) or link.job_id or "") return { "version": "short", "job_ref": job_ref, "channel": link.channel, "source": channel_to_source(link.channel), "tracking": link.tracking if isinstance(link.tracking, dict) else {}, "link_id": link.id, } def record_token_use(info: dict | None) -> None: """Count an application attributed to a short link. Best-effort only.""" if not info or not info.get("link_id"): return try: from django.db.models import F from django.utils import timezone from .models import JobTrackingLink JobTrackingLink.objects.filter(id=info["link_id"]).update( use_count=F("use_count") + 1, last_used_at=timezone.now() ) except Exception: # noqa: BLE001 — attribution stats never block an application logger.exception("Could not record tracking-link use") # --------------------------------------------------------------------------- # Token encode / decode # --------------------------------------------------------------------------- def make_tracking_token(job, channel: str, tracking: dict | None = None) -> str: """Encrypt (JD reference, channel, tracking metadata) into a URL-safe token. `job` may be a JobDescription or anything exposing `public_id`/`id`. `tracking` is optional free-form metadata (campaign, posting id, …) that comes back verbatim from resolve_tracking_token(). Trailing '=' padding is stripped — some clients mangle it in query strings; resolve_tracking_token() restores it, same as core.crypto. """ job_ref = str(getattr(job, "public_id", None) or getattr(job, "id", "") or "") payload = {"v": _VERSION, "j": job_ref, "s": (channel or "").upper()} if tracking: payload["t"] = tracking raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode() return _fernet().encrypt(raw).decode().rstrip("=") def _resolve_fernet_token(token: str) -> dict | None: """Decrypt a legacy Fernet token. None if missing/tampered/unreadable.""" padded = token + "=" * (-len(token) % 4) try: raw = _fernet().decrypt(padded.encode()).decode() payload = json.loads(raw) except (InvalidToken, ValueError, TypeError, UnicodeDecodeError): return None if not isinstance(payload, dict): return None channel = str(payload.get("s") or "").upper() if not channel: return None tracking = payload.get("t") return { "version": payload.get("v"), "job_ref": str(payload.get("j") or ""), "channel": channel, "source": channel_to_source(channel), "tracking": tracking if isinstance(tracking, dict) else {}, } def resolve_tracking_token(token: str) -> dict | None: """Validate a tracking token of either format and resolve what it stands for. Accepts the current short code and the legacy Fernet token — links already published in the long format keep working. Returns ``{"job_ref", "channel", "source", "tracking", "version"[, "link_id"]}``, or None when the token is missing, unknown, tampered with or unreadable. Never raises — an invalid token means "treat this as Direct". """ if not token or not isinstance(token, str): return None token = token.strip() if not token: return None # The prefix only decides which resolver to try first; both are attempted, # so a token is never rejected merely for not looking like its format. if token.startswith(_FERNET_PREFIX): return _resolve_fernet_token(token) or resolve_short_code(token) return resolve_short_code(token) or _resolve_fernet_token(token) # --------------------------------------------------------------------------- # URL building # --------------------------------------------------------------------------- def careers_base_url() -> str: """Public base URL of the careers site (same resolution as email deep links).""" base = (getattr(settings, "FRONTEND_URL", "") or "").rstrip("/") if base: return base origins = getattr(settings, "CORS_ALLOWED_ORIGINS", None) or ["http://localhost:3000"] return str(origins[0]).rstrip("/") def build_application_url(job, channel: str, tracking: dict | None = None, label: str = "", user=None) -> str: """Public application URL for a JD on one platform. The only query parameter is the opaque tracking token — the platform name never appears in the URL. Mints the short code (stable per job+channel, so previewing then publishing yields the same URL); if it can't be persisted, degrades to the self-contained Fernet token so a link is always produced. """ job_ref = getattr(job, "public_id", None) or getattr(job, "id", "") token = make_short_code(job, channel, tracking, label=label, user=user) if not token: token = make_tracking_token(job, channel, tracking) return f"{careers_base_url()}/careers/jobs/{job_ref}?t={token}" def resolve_source_for_application(token: str, job=None) -> tuple[str | None, dict | None]: """Resolve the candidate source for an application from a tracking token. Returns ``(source, info)``: - valid token -> (Candidate.SourceChoices value, resolved payload) - missing/invalid token -> (None, None), so the caller falls back to the existing Direct handling. When `job` is given, the token's JD reference must match it — a token minted for a different JD is ignored rather than trusted. """ info = resolve_tracking_token(token) if not info: return None, None if job is not None: job_ref = info.get("job_ref") or "" allowed = {str(getattr(job, "public_id", "") or ""), str(getattr(job, "id", "") or "")} if job_ref and job_ref not in allowed: logger.info( "Tracking token for JD %s presented on JD %s — ignoring.", job_ref, getattr(job, "id", "?"), ) return None, None return info["source"], info