IBM CP4D: Add Service Instance User Script

.env file

# CPD cluster admin credentials
# Copy this file to .env and fill in your values.

CPD_URL=cpd.apps.mycluster.example.com
CPD_ADMIN_USERNAME=admin
CPD_ADMIN_PASSWORD=your_admin_password_here

# Used by cpd_add_instance_user.py
CPD_INSTANCE_ID=your_service_instance_id (17....)
CPD_NEW_USERNAME=username_to_add
CPD_NEW_UID=uid_of_user (10003...)
CPD_NEW_ROLE=Admin

cpd_add_instance_user.py

#!/usr/bin/env python3
"""
CPD Add Service Instance User Script
Authenticates against CPD, then adds a user to a service instance
via POST /zen-data/v2/serviceInstance/users using a ZenApiKey token.
"""

import json
import os
import sys
import urllib.request
import urllib.error
import ssl
import getpass


# ---------------------------------------------------------------------------
# Shared helpers (same pattern as cpd_create_user.py)
# ---------------------------------------------------------------------------

def load_env(path: str = ".env") -> None:
    """Parse a .env file and populate os.environ (no external dependencies)."""
    try:
        with open(path) as fh:
            for line in fh:
                line = line.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                key, _, value = line.partition("=")
                value = value.strip().strip("\"'")
                os.environ.setdefault(key.strip(), value)
    except FileNotFoundError:
        pass  # .env is optional


def _ssl_ctx() -> ssl.SSLContext:
    """Return an SSL context with certificate verification disabled (mirrors curl -k)."""
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    return ctx


def prompt(label: str, secret: bool = False) -> str:
    """Prompt interactively; hide input when secret=True."""
    if secret:
        return getpass.getpass(f"{label}: ")
    value = input(f"{label}: ").strip()
    if not value:
        print(f"[ERROR] '{label}' cannot be empty.", file=sys.stderr)
        sys.exit(1)
    return value


# ---------------------------------------------------------------------------
# API calls
# ---------------------------------------------------------------------------

def get_bearer_token(cpd_url: str, admin_user: str, admin_password: str) -> str:
    """Authenticate and return the Bearer token."""
    url = f"https://{cpd_url}/icp4d-api/v1/authorize"
    payload = json.dumps({"username": admin_user, "password": admin_password}).encode()

    req = urllib.request.Request(
        url,
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST",
    )

    try:
        with urllib.request.urlopen(req, context=_ssl_ctx()) as resp:
            body = json.loads(resp.read().decode())
    except urllib.error.HTTPError as exc:
        print(f"[ERROR] Authentication failed ({exc.code}): {exc.read().decode()}", file=sys.stderr)
        sys.exit(1)

    token = body.get("token")
    if not token:
        print(f"[ERROR] No token in response: {body}", file=sys.stderr)
        sys.exit(1)

    return token


def list_instance_users(cpd_url: str, zen_token: str, instance_id: str) -> list:
    """List all users of a CPD service instance."""
    url = f"https://{cpd_url}/zen-data/v2/serviceInstance/users?sID={instance_id}"

    req = urllib.request.Request(
        url,
        headers={
            "Authorization": f"Bearer {zen_token}",
            "Accept": "application/json",
        },
        method="GET",
    )

    try:
        with urllib.request.urlopen(req, context=_ssl_ctx()) as resp:
            body = json.loads(resp.read().decode())
    except urllib.error.HTTPError as exc:
        print(f"[ERROR] List instance users failed ({exc.code}): {exc.read().decode()}", file=sys.stderr)
        sys.exit(1)

    # The API may return {"users": [...]} or a bare list or another structure.
    # Print raw response to help diagnose unexpected shapes.
    if "users" not in body:
        print(f"      [DEBUG] Raw response: {json.dumps(body, indent=2)}", file=sys.stderr)

    users = body.get("users", [])
    # Guard: keep only dict entries so we never crash on unexpected types.
    return [u for u in users if isinstance(u, dict)]


def add_instance_user(
    cpd_url: str,
    zen_token: str,
    instance_id: str,
    username: str,
    uid: str,
    role: str,
) -> dict:
    """Add a user to a CPD service instance and return the API response."""
    url = f"https://{cpd_url}/zen-data/v2/serviceInstance/users"
    payload = json.dumps(
        {
            "serviceInstanceID": instance_id,
            "users": [
                {
                    "role": role,
                    "uid": uid,
                    "username": username,
                }
            ],
        }
    ).encode()

    req = urllib.request.Request(
        url,
        data=payload,
        headers={
            "Authorization": f"Bearer {zen_token}",
            "Content-Type": "application/json",
        },
        method="POST",
    )

    try:
        with urllib.request.urlopen(req, context=_ssl_ctx()) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.HTTPError as exc:
        print(f"[ERROR] Add instance user failed ({exc.code}): {exc.read().decode()}", file=sys.stderr)
        sys.exit(1)


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

def main() -> None:
    load_env()
    print("=== IBM Cloud Pak for Data — Add Service Instance User ===\n")

    # --- Cluster / admin credentials (.env → interactive fallback) ---
    cpd_url    = os.environ.get("CPD_URL")            or prompt("CPD Route URL (without https://)")
    admin_user = os.environ.get("CPD_ADMIN_USERNAME") or (input("CPD admin username [admin]: ").strip() or "admin")
    admin_pass = os.environ.get("CPD_ADMIN_PASSWORD") or prompt("CPD admin password", secret=True)

    # --- Step 1: fetch Bearer / ZenApiKey token ---
    print("\n[1/2] Fetching Bearer token...")
    token = get_bearer_token(cpd_url, admin_user, admin_pass)
    print(f"      Token obtained (first 40 chars): {token[:40]}...")

    # --- Service instance user details (required from environment / .env) ---
    instance_id = os.environ.get("CPD_INSTANCE_ID") or prompt("Service Instance ID")

    def require_env(var: str) -> str:
        value = os.environ.get(var, "").strip()
        if not value:
            print(f"[ERROR] Environment variable '{var}' is required but not set.", file=sys.stderr)
            sys.exit(1)
        return value

    # --- Step 2: list current users ---
    print("\n[2/3] Current users in service instance...")
    current_users = list_instance_users(cpd_url, token, instance_id)
    if current_users:
        print(f"      {'USERNAME':<30} {'UID':<20} ROLE")
        print(f"      {'-'*30} {'-'*20} {'-'*20}")
        for u in current_users:
            print(f"      {u.get('username', u):<30} {str(u.get('uid', '')):<20} {u.get('role', '')}")
    else:
        print("      (no users found)")

    username = os.environ.get("CPD_NEW_USERNAME", "").strip()
    uid      = require_env("CPD_NEW_UID")
    role     = require_env("CPD_NEW_ROLE")

    # --- Step 3: add user to instance (skip if CPD_NEW_USERNAME is empty) ---
    if not username:
        print("\n[3/3] CPD_NEW_USERNAME not set — skipping add user step.")
        return

    print("\n[3/3] Adding user to service instance...")
    result = add_instance_user(
        cpd_url=cpd_url,
        zen_token=token,
        instance_id=instance_id,
        username=username,
        uid=uid,
        role=role,
    )

    print("\n[OK] User added to service instance successfully.")
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()