将 ONTAP 模式池加入 Active Directory

 ontap_execute "vserver show -fields vserver,operational-state"

 ontap_execute "network interface show -role data"

 ontap_execute "network ping -vserver SVM_NAME -destination DNS_OR_DC_IP -lif SVM_NAME-san-1"

 ontap_execute "vserver services name-service dns create -vserver SVM_NAME -domains AD_FQDN -name-servers AD_DNS_SERVERS"

 ontap_execute "vserver services name-service ns-switch modify -vserver SVM_NAME -database hosts -sources dns,files"

 ontap_execute "vserver services name-service dns show -vserver SVM_NAME"

 vserver cifs create -vserver SVM_NAME -cifs-server COMPUTER_NAME -domain AD_FQDN

 #!/usr/bin/env python3
"""Join an ONTAP-mode SVM to Active Directory by creating a CIFS server."""

import json
import os
import subprocess
import sys
import time
import urllib.error
import urllib.request

# Required environment variables.
REQUIRED = (
    "PROJECT_ID", "LOCATION", "POOL_NAME", "SVM_NAME",
    "AD_FQDN", "COMPUTER_NAME", "AD_USERNAME", "AD_PASSWORD",
)
missing = [name for name in REQUIRED if not os.environ.get(name)]
if missing:
    sys.exit(f"Set required environment variables: {', '.join(missing)}")

PROJECT_ID = os.environ["PROJECT_ID"]
LOCATION = os.environ["LOCATION"]
POOL_NAME = os.environ["POOL_NAME"]
SVM_NAME = os.environ["SVM_NAME"]
AD_FQDN = os.environ["AD_FQDN"]
COMPUTER_NAME = os.environ["COMPUTER_NAME"]
AD_USERNAME = os.environ["AD_USERNAME"]
AD_PASSWORD = os.environ["AD_PASSWORD"]

# Optional environment variables.
OU = os.environ.get("OU", "")
AD_SITE = os.environ.get("AD_SITE", "")
POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL", "10"))

# Kerberos encryption types advertised to Active Directory. AES-256 only by default.
# To add aes-128 or des (not recommended), extend this list before you run the script.
ADVERTISED_KDC_ENCRYPTIONS = ["aes-256"]

API_ROOT = (
    f"https://netapp.googleapis.com/v1/projects/{PROJECT_ID}"
    f"/locations/{LOCATION}/storagePools/{POOL_NAME}/ontap/api"
)

def get_access_token() -> str:
    return subprocess.check_output(
        ["gcloud", "auth", "print-access-token"], text=True
    ).strip()

def api_request(method: str, path: str, body=None) -> dict:
    headers = {
        "Authorization": f"Bearer {get_access_token()}",
        "Content-Type": "application/json",
    }
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(f"{API_ROOT}{path}", data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode()
        sys.exit(f"API request failed ({exc.code}): {detail}")

def job_record(response: dict) -> dict:
    return response.get("rawResponse") or response.get("body") or response

def job_state(response: dict) -> str:
    return job_record(response).get("state", "")

def format_job_error(response: dict) -> str:
    record = job_record(response)
    message = record.get("message") or record.get("error", {}).get("message")
    if message:
        return message
    return json.dumps(response, indent=2)

def wait_for_job(job_uuid: str) -> None:
    print(f"ONTAP job UUID: {job_uuid}")
    while True:
        response = api_request("GET", f"/cluster/jobs/{job_uuid}")
        state = job_state(response)
        print(f"  Job state: {state}")
        if state == "success":
            print("Domain join completed.")
            return
        if state in ("failure", "error"):
            sys.exit(f"Job failed: {format_job_error(response)}")
        time.sleep(POLL_INTERVAL)

def main() -> None:
    ad_domain = {
        "fqdn": AD_FQDN,
        "user": AD_USERNAME,
        "password": AD_PASSWORD,
    }
    if OU:
        ad_domain["organizational_unit"] = OU
    if AD_SITE:
        ad_domain["default_site"] = AD_SITE

    body = {
        "body": {
            "svm": {"name": SVM_NAME},
            "name": COMPUTER_NAME,
            "ad_domain": ad_domain,
            "security": {
                "advertised_kdc_encryptions": ADVERTISED_KDC_ENCRYPTIONS,
            },
        }
    }

    print(f"Creating CIFS server and joining {AD_FQDN}...")
    response = api_request("POST", "/protocols/cifs/services?return_records=true", body)

    job = (response.get("body") or {}).get("job") or response.get("job")
    if not job or not job.get("uuid"):
        print(json.dumps(response, indent=2))
        return
    wait_for_job(job["uuid"])

if __name__ == "__main__":
    main()

 ontap_execute "vserver cifs users-and-groups local-group add-members -vserver SVM_NAME -group-name BUILTIN\\Administrators -member-names NETBIOS_DOMAIN\\username"

 ontap_execute "vserver services name-service dns dynamic-update modify -vserver SVM_NAME -is-enabled true -use-secure true -vserver-fqdn COMPUTER_NAME.AD_FQDN"

 ontap_execute "vserver cifs show -vserver SVM_NAME"