fatfree.py

fatfree.py

import argparse
import os
import sys
import json
import httpx
import google.auth
import google.auth.transport.requests

# ------------------------------------------------------------------------------
# Configuration & Constants
# ------------------------------------------------------------------------------

API_BASE = "https://cloudidentityscim.googleapis.com/v1"
SCOPES = [
    "https://www.googleapis.com/auth/cloud-identity.groups",
    "https://www.googleapis.com/auth/admin.directory.user",
    "https://www.googleapis.com/auth/directory.user",
]

# SCIM Schema Constants
SCHEMA_USER = "urn:ietf:params:scim:schemas:core:2.0:User"
SCHEMA_USER_ENTERPRISE = "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User"
# Manually added to support orgUnitPath request (not in original discovery file)
#SCHEMA_USER_GOOGLE = "urn:ietf:params:scim:schemas:extension:google:2.0:User"
SCHEMA_USER_GOOGLE = 'urn:ietf:params:scim:schemas:extension:google:2.0:CloudIdentityUser'

SCHEMA_GROUP = "urn:ietf:params:scim:schemas:core:2.0:Group"
SCHEMA_GROUP_GOOGLE = "urn:ietf:params:scim:schemas:extension:google:2.0:CloudIdentityGroup"
SCHEMA_PATCH_OP = "urn:ietf:params:scim:api:messages:2.0:PatchOp"

# ------------------------------------------------------------------------------
# Logging & Debugging
# ------------------------------------------------------------------------------

def log_request(request):
    print(f"\n>> REQUEST: {request.method} {request.url}")
    print(">> Headers:")
    for k, v in request.headers.items():
        print(f"   {k}: {v}")
    if request.content:
        print(">> Body:")
        try:
            print(json.dumps(json.loads(request.content), indent=2))
        except:
            print(request.content.decode('utf-8', errors='replace'))
    print("-" * 40)

def log_response(response):
    response.read()
    print(f"\n<< RESPONSE: {response.status_code} {response.reason_phrase}")
    print("<< Headers:")
    for k, v in response.headers.items():
        print(f"   {k}: {v}")
    if response.content:
        print("<< Body:")
        try:
            print(json.dumps(json.loads(response.content), indent=2))
        except:
            print(response.content.decode('utf-8', errors='replace'))
    print("=" * 60 + "\n")

# ------------------------------------------------------------------------------
# Authentication
# ------------------------------------------------------------------------------

def get_authenticated_client(debug=False):
    """
    Obtains Google Default Credentials, refreshes the token, and returns
    an httpx.Client with the Authorization header set.
    """
    try:
        creds, project = google.auth.default(scopes=SCOPES)
        auth_req = google.auth.transport.requests.Request()
        creds.refresh(auth_req)
        
        headers = {
            "Authorization": f"Bearer {creds.token}",    
            "Content-Type": "application/json",
            "Accept": "application/json"
        }
        
        event_hooks = {}
        if debug:
            event_hooks = {'request': [log_request], 'response': [log_response]}

        return httpx.Client(headers=headers, timeout=30.0, event_hooks=event_hooks)
    except Exception as e:
        print(f"Error obtaining credentials: {e}")
        sys.exit(1)

# ------------------------------------------------------------------------------
# Helper Functions
# ------------------------------------------------------------------------------

def get_base_url(customer_id):
    """Constructs the versioned API base URL for a specific customer."""
    return f"{API_BASE}/customers/{customer_id}/v2"

def print_response(response: httpx.Response):
    """Pretty prints the API response."""
    if response.status_code == 204:
        print(f"Success: {response.status_code} No Content")
        return

    try:
        data = response.json()
        print(json.dumps(data, indent=2))
    except json.JSONDecodeError:
        print(f"Response ({response.status_code}): {response.text}")

    if response.is_error:
        sys.exit(1)

# ------------------------------------------------------------------------------
# Command Implementations: Users
# ------------------------------------------------------------------------------

def users_create(args, client, base_url):
    url = f"{base_url}/Users"
    
    payload = {
        "schemas": [SCHEMA_USER],
        "userName": args.email,
        "name": {
            "givenName": args.givenname,
            "familyName": args.lastname
        },
        "emails": [
            {
                "value": args.email,
                "type": "work",
                "primary": True
            }
        ],
        "active": args.active
    }

    if args.display_name:
        payload["displayName"] = args.display_name

    if args.secondary_email:
        for email in args.secondary_email:
            payload['emails'].append({'value': email, 'type': 'work'})
    if args.external_id:
        payload["externalId"] = args.external_id

    # Enterprise Extension
    enterprise_ext = {}
    if args.employee_number:
        enterprise_ext["employeeNumber"] = args.employee_number
    if args.cost_center:
        enterprise_ext["costCenter"] = args.cost_center
    if args.organization:
        enterprise_ext["organization"] = args.organization
    if args.department:
        enterprise_ext["department"] = args.department
    if args.manager_id:
         enterprise_ext["manager"] = {"value": args.manager_id}

    if enterprise_ext:
        payload["schemas"].append(SCHEMA_USER_ENTERPRISE)
        payload[SCHEMA_USER_ENTERPRISE] = enterprise_ext

    # Google User Extension (Org Unit Path)
    google_ext = {}
    if args.org_unit_path:
        google_ext["organizationalUnitPath"] = args.org_unit_path

    if google_ext:
        payload["schemas"].append(SCHEMA_USER_GOOGLE)
        payload[SCHEMA_USER_GOOGLE] = google_ext

    response = client.post(url, json=payload)
    print_response(response)

def users_delete(args, client, base_url):
    url = f"{base_url}/Users/{args.user_id}"
    response = client.delete(url)
    print_response(response)

def users_get(args, client, base_url):
    url = f"{base_url}/Users/{args.user_id}"
    params = {}
    if args.excluded_attributes:
        params['excludedAttributes'] = args.excluded_attributes
    response = client.get(url, params=params)
    print_response(response)

def users_list(args, client, base_url):
    url = f"{base_url}/Users"
    params = {}
    if args.filter:
        params['filter'] = args.filter
    response = client.get(url, params=params)
    print_response(response)

def users_patch(args, client, base_url):
    url = f"{base_url}/Users/{args.user_id}"
    
    # Priority 1: Raw JSON Body
    if args.json_body:
        try:
            body = json.loads(args.json_body)
        except json.JSONDecodeError:
            print("Error: Invalid JSON string provided.")
            sys.exit(1)
    else:
        # Priority 2: Construct Patch from Flags
        operations = []

        # Core User Schema Mappings
        if args.email:
            operations.append({
                "op": "replace",
                "path": "userName",
                "value": args.email
            })
        
        if args.givenname:
            operations.append({
                "op": "replace",
                "path": "name.givenName",
                "value": args.givenname
            })
        
        if args.lastname:
            operations.append({
                "op": "replace",
                "path": "name.familyName",
                "value": args.lastname
            })
            
        if args.active is not None: 
            operations.append({
                "op": "replace",
                "path": "active",
                "value": args.active
            })
            
        if args.display_name:
            operations.append({
                "op": "replace",
                "path": "displayName",
                "value": args.display_name
            })
            
        if args.external_id:
            operations.append({
                "op": "replace",
                "path": "externalId",
                "value": args.external_id
            })

        # Enterprise Extension Mappings
        urn_ent = SCHEMA_USER_ENTERPRISE
        
        if args.employee_number:
            operations.append({
                "op": "replace",
                "path": f"{urn_ent}:employeeNumber",
                "value": args.employee_number
            })
            
        if args.cost_center:
            operations.append({
                "op": "replace",
                "path": f"{urn_ent}:costCenter",
                "value": args.cost_center
            })
            
        if args.organization:
            operations.append({
                "op": "replace",
                "path": f"{urn_ent}:organization",
                "value": args.organization
            })
            
        if args.department:
            operations.append({
                "op": "replace",
                "path": f"{urn_ent}:department",
                "value": args.department
            })
            
        if args.manager_id:
            operations.append({
                "op": "replace",
                "path": f"{urn_ent}:manager.value",
                "value": args.manager_id
            })

        # Google Extension Mappings (Org Unit Path)
        urn_google = SCHEMA_USER_GOOGLE

        if args.org_unit_path:
            operations.append({
                "op": "replace",
                "path": f"{urn_google}:organizationalUnitPath",
                "value": args.org_unit_path
            })

        if not operations:
            print("Error: No updates specified. Please provide at least one field to update or use --json-body.")
            sys.exit(1)

        body = {
            "schemas": [SCHEMA_PATCH_OP],
            "Operations": operations
        }

    response = client.patch(url, json=body)
    print_response(response)

# ------------------------------------------------------------------------------
# Command Implementations: Groups
# ------------------------------------------------------------------------------

def groups_create(args, client, base_url):
    url = f"{base_url}/Groups"
    
    payload = {
        "schemas": [SCHEMA_GROUP, SCHEMA_GROUP_GOOGLE],
        "displayName": args.display_name,
        SCHEMA_GROUP_GOOGLE: {
            "email": args.email
        }
    }
    
    if args.description:
        payload[SCHEMA_GROUP_GOOGLE]["description"] = args.description

    if args.external_id:
        payload["externalId"] = args.external_id

    if args.members:
        member_list = [m.strip() for m in args.members.split(",") if m.strip()]
        if member_list:
            payload["members"] = [{"value": m} for m in member_list]

    response = client.post(url, json=payload)
    print_response(response)

def groups_delete(args, client, base_url):
    url = f"{base_url}/Groups/{args.group_id}"
    response = client.delete(url)
    print_response(response)

def groups_get(args, client, base_url):
    url = f"{base_url}/Groups/{args.group_id}"
    params = {}
    
    if args.attributes:
        params['attributes'] = args.attributes
        
    if args.excluded_attributes:
        params['excludedAttributes'] = args.excluded_attributes
        
    response = client.get(url, params=params)
    print_response(response)

def groups_list(args, client, base_url):
    url = f"{base_url}/Groups"
    params = {}
    if args.filter:
        params['filter'] = args.filter
    response = client.get(url, params=params)
    print_response(response)

def groups_patch(args, client, base_url):
    url = f"{base_url}/Groups/{args.group_id}"
    
    if args.json_body:
        try:
            body = json.loads(args.json_body)
        except json.JSONDecodeError:
            print("Error: Invalid JSON string provided.")
            sys.exit(1)
    else:
        operations = []
        if args.external_id:
            operations.append({
                "op": "replace",
                "path": "externalId",
                "value": args.external_id
            })
        if args.members:
            member_list = [m.strip() for m in args.members.split(",") if m.strip()]
            operations.append({
                "op": "replace",
                "path": "members",
                "value": [{"value": m} for m in member_list]
            })

        if not operations:
            print("Error: Must provide --json-body OR flags (--external-id, --members) to patch.")
            sys.exit(1)
            
        body = {
            "schemas": [SCHEMA_PATCH_OP],
            "Operations": operations
        }

    response = client.patch(url, json=body)
    print_response(response)

# ------------------------------------------------------------------------------
# Command Implementations: Schemas
# ------------------------------------------------------------------------------

def schemas_list(args, client, base_url):
    url = f"{base_url}/Schemas"
    response = client.get(url)
    print_response(response)

def schemas_get(args, client, base_url):
    url = f"{base_url}/Schemas/{args.schema_id}"
    response = client.get(url)
    print_response(response)

# ------------------------------------------------------------------------------
# Main Argument Parser
# ------------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(description="Google Cloud Identity SCIM API CLI")
    
    # Global arguments
    parser.add_argument(
        "--customer-id", 
        help="The Customer ID (e.g., C0123456). Defaults to env var GOOGLE_CUSTOMER_ID.",
        default=os.environ.get("GOOGLE_CUSTOMER_ID")
    )
    parser.add_argument("--debug", action="store_true", help="Enable verbose HTTP logging")

    subparsers = parser.add_subparsers(dest="resource", help="Resource to manage")
    subparsers.required = True

    # --- Users Resource ---
    users_parser = subparsers.add_parser("users.create", help="Create a new user")
    users_parser.add_argument("--email", required=True, help="Primary email/username")
    users_parser.add_argument("--givenname", required=True, help="Given name")
    users_parser.add_argument("--lastname", required=True, help="Family name")
    users_parser.add_argument("--active", type=bool, default=True, help="Account active status")
    users_parser.add_argument('--secondary-email', nargs='*', help="Secondary emails.")
    users_parser.add_argument("--external-id", help="External ID for the user")
    users_parser.add_argument("--display-name", help="Display name for the user")
    users_parser.add_argument("--org-unit-path", help="The Org Unit Path (e.g. /Corp/Engineering)")
    users_parser.add_argument("--employee-number", help="Employee Number")
    users_parser.add_argument("--cost-center", help="Cost Center")
    users_parser.add_argument("--organization", help="Organization")
    users_parser.add_argument("--department", help="Department")
    users_parser.add_argument("--manager-id", help="The SCIM ID of the user's manager")
    users_parser.set_defaults(func=users_create)

    users_del_parser = subparsers.add_parser("users.delete", help="Delete a user")
    users_del_parser.add_argument("user_id", help="The unique User ID")
    users_del_parser.set_defaults(func=users_delete)

    users_get_parser = subparsers.add_parser("users.get", help="Get a user")
    users_get_parser.add_argument("user_id", help="The unique User ID")
    users_get_parser.add_argument("--excluded-attributes", help="Comma-separated list of attributes to exclude")
    users_get_parser.set_defaults(func=users_get)

    users_list_parser = subparsers.add_parser("users.list", help="List users")
    users_list_parser.add_argument("--filter", help="SCIM filter string")
    users_list_parser.set_defaults(func=users_list)

    # users.patch
    users_patch_parser = subparsers.add_parser("users.patch", help="Patch a user")
    users_patch_parser.add_argument("user_id", help="The unique User ID")
    users_patch_parser.add_argument("--json-body", help="Raw JSON body (optional)")
    
    # Patch flags (Core)
    users_patch_parser.add_argument("--email", help="Update Primary email/username")
    users_patch_parser.add_argument("--givenname", help="Update Given name")
    users_patch_parser.add_argument("--lastname", help="Update Family name")
    users_patch_parser.add_argument("--active", type=lambda x: (str(x).lower() == 'true'), help="Update Active status (true/false)")
    users_patch_parser.add_argument("--external-id", help="Update External ID")
    users_patch_parser.add_argument("--display-name", help="Update Display name")
    
    # Patch flags (Enterprise)
    users_patch_parser.add_argument("--employee-number", help="Update Employee Number")
    users_patch_parser.add_argument("--cost-center", help="Update Cost Center")
    users_patch_parser.add_argument("--organization", help="Update Organization")
    users_patch_parser.add_argument("--department", help="Update Department")
    users_patch_parser.add_argument("--manager-id", help="Update Manager ID")
    
    # Patch flags (Google)
    users_patch_parser.add_argument("--org-unit-path", help="Update Org Unit Path")
    
    users_patch_parser.set_defaults(func=users_patch)

    # --- Groups Resource ---
    groups_create_parser = subparsers.add_parser("groups.create", help="Create a new group")
    groups_create_parser.add_argument("--display-name", required=True, help="Group display name")
    groups_create_parser.add_argument("--email", required=True, help="Group email address")
    groups_create_parser.add_argument("--description", help="Group description")
    groups_create_parser.add_argument("--external-id", help="Group External ID")
    groups_create_parser.add_argument("--members", help="Comma-separated list of Member IDs")
    groups_create_parser.set_defaults(func=groups_create)

    groups_del_parser = subparsers.add_parser("groups.delete", help="Delete a group")
    groups_del_parser.add_argument("group_id", help="The unique Group ID")
    groups_del_parser.set_defaults(func=groups_delete)

    groups_get_parser = subparsers.add_parser("groups.get", help="Get a group")
    groups_get_parser.add_argument("group_id", help="The unique Group ID")
    groups_get_parser.add_argument("--attributes", help="Comma-separated list of attributes to return")
    groups_get_parser.add_argument("--excluded-attributes", help="Comma-separated list to exclude")
    groups_get_parser.set_defaults(func=groups_get)

    groups_list_parser = subparsers.add_parser("groups.list", help="List groups")
    groups_list_parser.add_argument("--filter", help="SCIM filter string")
    groups_list_parser.set_defaults(func=groups_list)

    groups_patch_parser = subparsers.add_parser("groups.patch", help="Patch a group")
    groups_patch_parser.add_argument("group_id", help="The unique Group ID")
    groups_patch_parser.add_argument("--json-body", help="Raw JSON body")
    groups_patch_parser.add_argument("--external-id", help="Set External ID (replace op)")
    groups_patch_parser.add_argument("--members", help="Set Members List (replace op)")
    groups_patch_parser.set_defaults(func=groups_patch)

    # --- Schemas Resource ---
    schemas_list_parser = subparsers.add_parser("schemas.list", help="List schemas")
    schemas_list_parser.set_defaults(func=schemas_list)

    schemas_get_parser = subparsers.add_parser("schemas.get", help="Get a schema")
    schemas_get_parser.add_argument("schema_id", help="The Schema ID")
    schemas_get_parser.set_defaults(func=schemas_get)

    # Parse and Execute
    args = parser.parse_args()
    
    if not args.customer_id:
        print("Error: --customer-id is required (or set GOOGLE_CUSTOMER_ID env var).")
        sys.exit(1)

    client = get_authenticated_client(debug=args.debug)
    base_url = get_base_url(args.customer_id)
    
    try:
        args.func(args, client, base_url)
    except httpx.HTTPError as h:
        print(f"HTTP Error: {h}")
    except Exception as e:
        print(f"Unexpected Error: {e}")

if __name__ == "__main__":
    main()