Add script to sync Jamf user assignments to Fleet device mapping (#42177)
Adds `docs/solutions/api-scripts/sync_jamf_users_to_fleet.py`, a Python script that: - Scrapes every computer in Jamf Pro via the Classic API (`/JSSResource/computers/subset/basic`) - Pulls the assigned user's email (falls back to username if no email is set) - Matches each device to a Fleet host by serial number - Assigns that user to the host in Fleet via `PUT /api/v1/fleet/hosts/:id/device_mapping` Features: - Supports both OAuth 2.0 client credentials and username/password auth for Jamf - `--dry-run` flag to preview changes without modifying Fleet - Summary statistics on completion (assigned, skipped, errors) Resolves fleetdm/confidential#14972
This commit is contained in:
kitzy
@@ -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()
|
||||
Reference in New Issue
Block a user