diff --git a/docs/solutions/api-scripts/sync_jamf_users_to_fleet.py b/docs/solutions/api-scripts/sync_jamf_users_to_fleet.py new file mode 100644 index 0000000000..74a5b728ef --- /dev/null +++ b/docs/solutions/api-scripts/sync_jamf_users_to_fleet.py @@ -0,0 +1,307 @@ +#!/usr/bin/env python3 +"""Sync device-to-user assignments from Jamf Pro to Fleet. + +For each computer in Jamf this script: + 1. Reads the assigned user's email (falls back to username if no email is set). + 2. Finds the matching host in Fleet by serial number. + 3. Sets the human-device mapping email on that Fleet host via the API. + +Required environment variables +------------------------------- + JAMF_URL Base URL of your Jamf Pro server + (e.g. https://org.jamfcloud.com) + + --- Jamf auth: choose ONE of the following two options --- + + Option A – OAuth 2.0 client credentials (Jamf Pro 10.49+, recommended): + JAMF_CLIENT_ID Jamf Pro API client ID + JAMF_CLIENT_SECRET Jamf Pro API client secret + + Option B – username / password: + JAMF_USERNAME Jamf Pro username + JAMF_PASSWORD Jamf Pro password + + --- Fleet --- + FLEET_URL Base URL of your Fleet server + (e.g. https://fleet.example.com) + FLEET_API_TOKEN Fleet API token (Settings → My account → Get API token) + +Usage +----- + # 1. Install dependencies (one-time) + pip install requests + + # 2. Export credentials + export JAMF_URL="https://org.jamfcloud.com" + + # Option A – OAuth (Jamf Pro 10.49+, recommended) + export JAMF_CLIENT_ID="your-client-id" + export JAMF_CLIENT_SECRET="your-client-secret" + + # Option B – username / password + export JAMF_USERNAME="your-jamf-username" + export JAMF_PASSWORD="your-jamf-password" + + export FLEET_URL="https://fleet.example.com" + export FLEET_API_TOKEN="your-fleet-api-token" + + # 3. Dry run first to preview changes without modifying Fleet + python3 sync_jamf_users_to_fleet.py --dry-run + + # 4. Run for real + python3 sync_jamf_users_to_fleet.py +""" + +import argparse +import os +import sys +import time + +import requests + + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + +JAMF_URL = os.getenv("JAMF_URL", "").rstrip("/") +JAMF_CLIENT_ID = os.getenv("JAMF_CLIENT_ID", "") +JAMF_CLIENT_SECRET = os.getenv("JAMF_CLIENT_SECRET", "") +JAMF_USERNAME = os.getenv("JAMF_USERNAME", "") +JAMF_PASSWORD = os.getenv("JAMF_PASSWORD", "") + +FLEET_URL = os.getenv("FLEET_URL", "").rstrip("/") +FLEET_API_TOKEN = os.getenv("FLEET_API_TOKEN", "") + +# Small delay between Fleet write calls to avoid hitting rate limits +FLEET_REQUEST_DELAY = 0.05 # seconds + + +# --------------------------------------------------------------------------- +# Startup validation +# --------------------------------------------------------------------------- + +def _die(msg): + print(f"ERROR: {msg}", file=sys.stderr) + sys.exit(1) + + +if not JAMF_URL: + _die("JAMF_URL is not set.") + +if not FLEET_URL: + _die("FLEET_URL is not set.") + +if not FLEET_API_TOKEN: + _die("FLEET_API_TOKEN is not set.") + +_have_oauth = bool(JAMF_CLIENT_ID and JAMF_CLIENT_SECRET) +_have_basic = bool(JAMF_USERNAME and JAMF_PASSWORD) + +if not _have_oauth and not _have_basic: + _die( + "Jamf credentials are not set. " + "Provide JAMF_CLIENT_ID + JAMF_CLIENT_SECRET (preferred) " + "or JAMF_USERNAME + JAMF_PASSWORD." + ) + + +# --------------------------------------------------------------------------- +# Jamf authentication +# --------------------------------------------------------------------------- + +_jamf_token: str = "" +_jamf_token_expires_at: float = 0.0 + + +def _refresh_jamf_token() -> str: + """Obtain or renew a Jamf Pro bearer token and return it.""" + global _jamf_token, _jamf_token_expires_at + + if _have_oauth: + # OAuth 2.0 client credentials (Jamf Pro 10.49+) + resp = requests.post( + f"{JAMF_URL}/api/oauth/token", + data={ + "grant_type": "client_credentials", + "client_id": JAMF_CLIENT_ID, + "client_secret": JAMF_CLIENT_SECRET, + }, + headers={"Accept": "application/json"}, + timeout=30, + ) + resp.raise_for_status() + body = resp.json() + _jamf_token = body["access_token"] + _jamf_token_expires_at = time.time() + body.get("expires_in", 1800) + else: + # Basic-auth token exchange (Jamf Pro 10.35+) + resp = requests.post( + f"{JAMF_URL}/api/v1/auth/token", + auth=(JAMF_USERNAME, JAMF_PASSWORD), + headers={"Accept": "application/json"}, + timeout=30, + ) + resp.raise_for_status() + body = resp.json() + _jamf_token = body["token"] + _jamf_token_expires_at = time.time() + 1800 # default 30-min lifetime + + return _jamf_token + + +def _jamf_headers() -> dict: + """Return request headers with a valid Jamf bearer token.""" + global _jamf_token, _jamf_token_expires_at + # Refresh 60 s before expiry to avoid mid-run failures + if not _jamf_token or time.time() >= _jamf_token_expires_at - 60: + _refresh_jamf_token() + return { + "Authorization": f"Bearer {_jamf_token}", + "Accept": "application/json", + } + + +# --------------------------------------------------------------------------- +# Jamf helpers +# --------------------------------------------------------------------------- + +def get_all_jamf_computers(): + """Yield computer records from Jamf (id + serial_number + user). + + Uses the Classic API /subset/basic endpoint, which returns serial_number, + email_address, and username for every computer in a single request — + avoiding a separate per-device location fetch. + """ + resp = requests.get( + f"{JAMF_URL}/JSSResource/computers/subset/basic", + headers=_jamf_headers(), + timeout=60, + ) + resp.raise_for_status() + computers = resp.json().get("computers", []) + + for c in computers: + email = (c.get("email_address") or "").strip() + username = (c.get("username") or "").strip() + yield { + "jamf_id": c["id"], + "serial_number": (c.get("serial_number") or "").strip(), + "user": email or username or None, + } + + +# --------------------------------------------------------------------------- +# Fleet helpers +# --------------------------------------------------------------------------- + +_fleet_headers = { + "Authorization": f"Bearer {FLEET_API_TOKEN}", + "Content-Type": "application/json", +} + + +def get_fleet_host_by_serial(serial: str) -> dict | None: + """Look up a Fleet host by serial number. Returns the host dict or None.""" + resp = requests.get( + f"{FLEET_URL}/api/v1/fleet/hosts/identifier/{serial}", + headers=_fleet_headers, + timeout=30, + ) + if resp.status_code == 404: + return None + resp.raise_for_status() + return resp.json().get("host") + + +def assign_fleet_device_mapping(host_id: int, email: str) -> None: + """Set the custom device-mapping email for a Fleet host.""" + resp = requests.put( + f"{FLEET_URL}/api/v1/fleet/hosts/{host_id}/device_mapping", + headers=_fleet_headers, + json={"email": email}, + timeout=30, + ) + resp.raise_for_status() + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + +def main(): + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument( + "--dry-run", + action="store_true", + help="Log what would be assigned without making any changes in Fleet.", + ) + args = parser.parse_args() + dry_run = args.dry_run + + if dry_run: + print("DRY RUN — no changes will be made in Fleet.") + print() + + assigned = 0 + skipped_no_serial = 0 + skipped_no_user = 0 + skipped_not_in_fleet = 0 + errors = 0 + + print(f"Fetching computers from Jamf Pro ({JAMF_URL})…") + + for computer in get_all_jamf_computers(): + serial = computer["serial_number"] + jamf_id = computer["jamf_id"] + + if not serial: + skipped_no_serial += 1 + continue + + user = computer["user"] + + if not user: + print(f" [SKIP] {serial}: no user assigned in Jamf") + skipped_no_user += 1 + continue + + # --- Look up the host in Fleet by serial number --- + try: + host = get_fleet_host_by_serial(serial) + except requests.HTTPError as exc: + print(f" [WARN] Fleet error looking up serial {serial}: {exc}") + errors += 1 + continue + + if host is None: + print(f" [SKIP] {serial}: not found in Fleet") + skipped_not_in_fleet += 1 + continue + + # --- Assign the user in Fleet --- + if dry_run: + print(f" [DRY] {serial} → {user} (Fleet host ID {host['id']})") + assigned += 1 + else: + try: + assign_fleet_device_mapping(host["id"], user) + print(f" [OK] {serial} → {user} (Fleet host ID {host['id']})") + assigned += 1 + except requests.HTTPError as exc: + print(f" [WARN] Fleet error assigning {user} to host {host['id']}: {exc}") + errors += 1 + + time.sleep(FLEET_REQUEST_DELAY) + + print() + print("Done (dry run — no changes were made)." if dry_run else "Done.") + print(f" {'Would assign' if dry_run else 'Assigned'} : {assigned}") + print(f" Skipped (no serial in Jamf): {skipped_no_serial}") + print(f" Skipped (no user in Jamf) : {skipped_no_user}") + print(f" Skipped (not in Fleet) : {skipped_not_in_fleet}") + print(f" Errors : {errors}") + + +if __name__ == "__main__": + main()