IBM CP4D: Add Service Instance User Script - Revision history

Ebasso: Created page with " == .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 ==
2026-09-11T14:39:21Z

Created page with " == .env file == <syntaxhighlight lang="properties"> # 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 </syntaxhighlight> == cpd_add_instance_user.py == <synta..."

New page

== .env file ==

<syntaxhighlight lang="properties">
# 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
</syntaxhighlight>

== cpd_add_instance_user.py ==
<syntaxhighlight lang="python">
#!/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()

</syntaxhighlight>

= Ver também =

[[Category:Cognos]]
[[Category:CP4D]]
[[Category:IBM Maximo]]
[[Category:IBM MAS]]