本ページでは、KeeperPAMの「PAMスクリプトのみ実行」オプションを使用して、Azureアプリケーションのシークレットをローテーションさせる方法について取り扱います。 これはPAMユーザーのローテーション設定内のオプションであり、ゲートウェイが主要なローテーション方法をスキップし、ボルト内のPAMユーザーレコードに添付されたポストローテーションスクリプトを直接実行するよう指示する設定です。
以下には、要件、手順、Pythonスクリプトの例を記載しています。スクリプトでは、以前のシークレットを削除して、新しいアプリケーションシークレットをKeeperに保存するなど、安全にアプリケーションシークレットをローテーションするようにします。この新しいシークレットは自動的に、既に許可されているすべてのKSMアプリケーションおよびユーザーが利用できるようになります。
pip3 install keeper_secrets_manager_core
pip3 install azure.identity
import logging
import requests
import sys
import base64
import json
from azure.identity import ClientSecretCredential
from datetime import datetime, timedelta, timezone
from keeper_secrets_manager_core import SecretsManager
from keeper_secrets_manager_core.storage import FileKeyValueStorage
# sys.stdinは配列ではなく、添字でアクセスできません（例：sys.stdin[0]）。
for base64_params in sys.stdin:
# ゲートウェイからパラメータを取得する
params = json.loads(base64.b64decode(base64_params).decode('utf-8'))
break
# ポストローテーションスクリプトに添付されたレコードをKeeperから取得する
records = json.loads(base64.b64decode(params.get('records')).decode('utf-8'))
# ゲートウェイ設定を使用してKeeperシークレットマネージャーSDKを初期化する
secrets_manager = SecretsManager(config=FileKeyValueStorage('/etc/keeper-gateway/gateway-config.json'))
# ログの設定
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
GRAPH_API_URL = "https://graph.microsoft.com/v1.0"
# ClientSecretCredentialを使用してMicrosoft Graphアクセストークンを取得する関数
def get_access_token(tenant_id, client_id, client_secret):
try:
credential = ClientSecretCredential(tenant_id=tenant_id, client_id=client_id, client_secret=client_secret)
token = credential.get_token("https://graph.microsoft.com/.default")
logger.info("Authentication successful.")
return token.token
except Exception as e:
logger.error(f"Authentication failed: {e}")
raise
# 新しいクライアントシークレットを作成し、そのIDとシークレットテキストを返す関数
def create_client_secret(access_token, target_app_id, secret_expiry_days=365):
try:
# タイムゾーンを考慮した日時を使用して、シークレットの有効期限を設定する
expiry_time = (datetime.now(timezone.utc) + timedelta(days=secret_expiry_days)).isoformat()
# 新しいクライアントシークレットを作成するためのAPIリクエストボディ
body = {
"passwordCredential": {
"displayName": "Client Secret Rotated by Keeper",
"endDateTime": expiry_time
}
}
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
response = requests.post(f"{GRAPH_API_URL}/applications/{target_app_id}/addPassword", json=body, headers=headers)
if response.status_code == 200:
response_json = response.json()
new_client_secret = response_json['secretText']
secret_id = response_json['keyId'] # 新しいクライアントシークレットのIDを取得
# 新しいシークレットとそのIDの両方をログに記録する
logger.info(f"New client secret created for Application {target_app_id}. Secret ID: {secret_id}.")
return secret_id, new_client_secret, expiry_time
else:
logger.error(f"Failed to create client secret: {response.status_code}, {response.text}")
raise Exception(f"Failed to create client secret: {response.status_code}")
except Exception as e:
logger.error(f"An error occurred: {e}")
raise
# 既存のクライアントシークレットを取得する関数
def get_existing_client_secrets(access_token, target_app_id):
try:
headers = {
"Authorization": f"Bearer {access_token}"
}
response = requests.get(f"{GRAPH_API_URL}/applications/{target_app_id}", headers=headers)
if response.status_code == 200:
app = response.json()
logger.info(f"Retrieved existing client secrets for Application {target_app_id}.")
return app.get('passwordCredentials', [])
else:
logger.error(f"Failed to retrieve client secrets: {response.status_code}, {response.text}")
raise Exception(f"Failed to retrieve client secrets: {response.status_code}")
except Exception as e:
logger.error(f"An error occurred: {e}")
raise
# 新しく作成されたクライアントシークレットを除く、古いクライアントシークレットを削除する関数
def delete_old_secrets(access_token, target_app_id, new_client_secret_id):
try:
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
existing_secrets = get_existing_client_secrets(access_token, target_app_id)
# すべてのシークレットを繰り返し処理し、新しく作成されたシークレットでないものを削除する
for secret in existing_secrets:
if secret['keyId'] != new_client_secret_id:
body = {
"keyId": secret['keyId']
}
response = requests.post(f"{GRAPH_API_URL}/applications/{target_app_id}/removePassword", json=body, headers=headers)
if response.status_code == 204:
logger.info(f"Deleted secret with ID: {secret['keyId']} for Application {target_app_id}.")
else:
logger.error(f"Failed to delete secret: {response.status_code}, {response.text}")
raise Exception(f"Failed to delete secret: {response.status_code}")
except Exception as e:
logger.error(f"An error occurred: {e}")
raise
# メイン関数
def rotate_client_secret(tenant_id, client_id, client_secret, target_app_id, secret_expiry_days=365, delete_old_secrets_flag=False):
try:
# ステップ1: アクセストークンを取得する
access_token = get_access_token(tenant_id, client_id, client_secret)
# ステップ2: 新しいクライアントシークレットを作成し、そのIDを取得する
new_client_secret_id, new_client_secret, expiry_time = create_client_secret(access_token, target_app_id, secret_expiry_days)
if not new_client_secret:
logger.error("Failed to create a new client secret. Exiting.")
return None, None, None # シークレットの作成に失敗した場合は、さらに実行を防止する
logger.info(f"New client secret id: {new_client_secret_id}. Expiration date: {expiry_time}")
# ステップ3: 必要に応じて古いシークレットを削除するが、新しく作成されたシークレットは除外する
if delete_old_secrets_flag:
delete_old_secrets(access_token, target_app_id, new_client_secret_id)
return new_client_secret_id, new_client_secret, expiry_time
except Exception as e:
logger.error(f"An error occurred during the secret rotation process: {e}")
return None, None, None # エラーが発生した場合は、`None`を返す
# ゲートウェイからの入力と、KSMを使用した完全なPAMユーザークエリ
pam_user_to_update = secrets_manager.get_secrets([params.get('userRecordUid')])[0]
# アプリケーションのオブジェクトIDを取得してローテーションを実行する
try:
target_app_id = pam_user_to_update.custom_field('application_object_id', single=True)
except Exception as e:
raise RuntimeError(f"No app object id found, the script will exit after this error: {e}")
# 管理者ユーザーアカウントのアクセスキーを取得する
admin_cred_provided = False
for record in records:
if "application_client_id" in record:
client_id = record["application_client_id"]
client_secret = record["client_secret"]
tenant_id = record["tenant_id"]
admin_cred_provided = True
logger.info(f"Admin creds provided. Using secret_id {client_id}, from attached record UID {record['uid']}.")
break
elif "clientId" in record:
client_id = record["clientId"]
client_secret = record["clientSecret"]
tenant_id = record["tenantId"]
admin_cred_provided = True
logger.info(f"Admin creds provided. Using secret_id {client_id}, from attached PAM Config record UID {record['uid']}.")
break
# もし添付されたレコードにアプリケーションキーが提供されていない場合は、PAM設定を使用する
if admin_cred_provided == False:
try:
pam_config = secrets_manager.get_secrets([params.get('providerRecordUid')])[0]
client_id = pam_config.field('clientId', single=True)
client_secret = pam_config.field('clientSecret', single=True)
tenant_id = pam_config.field('tenantId', single=True)
admin_cred_provided = True
logger.info(f"Using PAM config creds. Using secret_id {client_id}, from PAM Config UID.")
except Exception as e:
logger.error(f"Error retrieving admin creds from PAM config: {e}")
# 認証情報が提供されていない場合は終了する
if not admin_cred_provided:
raise RuntimeError("No admin creds provided. The script will exit.")
# クライアントシークレットをローテーションし、成功/失敗を処理する
new_client_secret_id, new_client_secret, expiry_time = rotate_client_secret(
tenant_id, client_id, client_secret, target_app_id, secret_expiry_days=365, delete_old_secrets_flag=True
)
if new_client_secret_id and new_client_secret:
logger.info("\nRotation Result:")
logger.info(f"New secret id: {new_client_secret_id}")
# KeeperでPAMユーザーレコードを更新する
try:
pam_user_to_update.custom_field('client_secret_id', new_client_secret_id)
pam_user_to_update.custom_field('client_secret', new_client_secret)
pam_user_to_update.custom_field('expires', expiry_time)
secrets_manager.save(pam_user_to_update)
except Exception as e:
logger.error(f"Error while updating PAM User record in Keeper: {e}")
