정책 나열 및 가져오기
이 페이지에서는 정책을 나열하고 가져오는 방법을 보여주는 코드 예시를 제공합니다.
시작하기 전에
정책 API 설정을 완료합니다.
정책 나열
다음 예에서는 Python을 사용하여 조직의 정책을 나열하는 방법을 보여줍니다.
"""Sample script to demonstrate the use of the List method in the Policy API."""
from collections.abc import Mapping, Sequence
import json
import pprint
import time
from typing import Any
import urllib.request
from absl import app
from absl import flags
import google.auth.transport.requests
from google.oauth2 import service_account
SCOPES = ['https://www.googleapis.com/auth/cloud-identity.policies']
BASE_URL = 'https://cloudidentity.googleapis.com/'
VERSIONED_BASE_URL = f'{BASE_URL}v1/'
_SA_FILE = flags.DEFINE_string(
name='sa_file',
default=None,
help='Service account credentials file path',
required=True,
)
_ADMIN_EMAIL = flags.DEFINE_string(
name='admin_email',
default=None,
help='Administrator email to call as',
required=True,
)
_PAGE_SIZE = flags.DEFINE_integer(
name='page_size',
default=50,
help='Page size for the List API request',
required=False,
)
# To list all policies, set FILTER to '';
# To list policies for a specific customer, set FILTER to
# 'customer == "customers/{obfuscated_target_customer_id}"';
# To list policies for a specific Application, such as Gmail, set FILTER to
# 'setting.type.matches("gmail.*")';
# To list policies for a specific Setting, such as service_status, set FILTER to
# 'setting.type.matches(".*service_status")'.
_LIST_FILTER = flags.DEFINE_string(
name='list_filter',
default='',
help='Filter for the List API request',
required=False,
)
def create_delegated_credentials(
sa_file: str, user_email: str
) -> service_account.Credentials:
"""Creates delegated credentials for the user.
Args:
sa_file: The service account credentials file path.
user_email: The administrator email to call as.
Returns:
The delegated credentials for the user.
"""
credentials = service_account.Credentials.from_service_account_file(
sa_file, scopes=SCOPES
)
delegated_credentials = credentials.with_subject(user_email)
return delegated_credentials
def build_list_policies_request(
page_size: int, list_filter: str, page_token: str, access_token: str
) -> urllib.request.Request:
"""Builds the request for the List Policies API.
Args:
page_size: The page size for the request.
list_filter: The filter for the request.
page_token: The page token for the request.
access_token: The access token for the API.
Returns:
The request for the List Policies API.
"""
list_url = (
f'{VERSIONED_BASE_URL}policies?page_size={page_size}'
f'&filter={list_filter}&page_token={page_token}'
)
request = urllib.request.Request(list_url)
request.add_header('Authorization', 'Bearer ' + access_token)
return request
def call_list_policies_api(
request: urllib.request.Request,
) -> Mapping[str, Any]:
"""Calls the List Policies API.
Args:
request: The request for the List Policies API.
Returns:
The response for the List Policies API.
"""
content = urllib.request.urlopen(request).read()
return json.loads(content)
def call_list_policies_api_till_last_page(
access_token: str, page_size: int, list_filter: str
) -> None:
"""Calls the List Policies API until the last page.
Args:
access_token: The access token for the API.
page_size: The page size for the request.
list_filter: The filter for the request.
"""
page_token = ''
# Paginate until the last page.
while True:
list_policies_request = build_list_policies_request(
page_size, list_filter, page_token, access_token
)
list_policies_response = call_list_policies_api(list_policies_request)
print_list_policies_response(list_policies_response)
if (
'nextPageToken' not in list_policies_response
or not list_policies_response['nextPageToken']
):
print('This is the last page.')
break
page_token = list_policies_response['nextPageToken']
time.sleep(1)
def print_list_policies_response(response: Mapping[str, Any]) -> None:
pp = pprint.PrettyPrinter(indent=4)
if 'policies' in response:
for policy in response['policies']:
pp.pprint(policy)
if 'nextPageToken' in response:
print('Next page token: ' + response['nextPageToken'])
def main(argv: Sequence[str]):
if len(argv) > 4:
raise app.UsageError('Too many command-line arguments.')
dc = create_delegated_credentials(_SA_FILE.value, _ADMIN_EMAIL.value)
dc.refresh(google.auth.transport.requests.Request())
call_list_policies_api_till_last_page(
dc.token,
_PAGE_SIZE.value,
_LIST_FILTER.value,
)
if __name__ == '__main__':
app.run(main)
정책 가져오기
다음 예에서는 Python을 사용하여 특정 정책을 가져오는 방법을 보여줍니다.
"""Sample script to demonstrate the use of the get method in the Policy API."""
from collections.abc import Sequence
import json
import pprint
import urllib.request
from absl import app
from absl import flags
import google.auth.transport.requests
from google.oauth2 import service_account
SCOPES = ['https://www.googleapis.com/auth/cloud-identity.policies']
BASE_URL = 'https://cloudidentity.googleapis.com/'
VERSIONED_BASE_URL = f'{BASE_URL}v1/'
_SA_FILE = flags.DEFINE_string(
name='sa_file',
default=None,
help='Service account credentials file path',
required=True,
)
_ADMIN_EMAIL = flags.DEFINE_string(
name='admin_email',
default=None,
help='Administrator email to call as',
required=True,
)
_POLICY_NAME = flags.DEFINE_string(
name='policy_name',
default=None,
help='Policy name of the policy to get',
required=True,
)
def create_delegated_credentials(
sa_file: str, user_email: str
) -> service_account.Credentials:
"""Creates delegated credentials for the user.
Args:
sa_file: The service account credentials file path.
user_email: The administrator email to call as.
Returns:
The delegated credentials for the user.
"""
credentials = service_account.Credentials.from_service_account_file(
sa_file, scopes=SCOPES
)
delegated_credentials = credentials.with_subject(user_email)
return delegated_credentials
def build_get_policy_request(
policy_name: str, access_token: str
) -> urllib.request.Request:
"""Builds the request for the Get Policy API.
Args:
policy_name: The policy name to get.
access_token: The access token for the API.
Returns:
The request for the Get Policy API.
"""
list_url = f'{VERSIONED_BASE_URL}{policy_name}'
request = urllib.request.Request(list_url)
request.add_header('Authorization', 'Bearer ' + access_token)
return request
def call_get_policy_api(access_token: str, policy_name: str) -> None:
"""Calls the Get Policy API.
Args:
access_token: The access token for the API.
policy_name: The policy name to get.
"""
request = build_get_policy_request(policy_name, access_token)
content = urllib.request.urlopen(request).read()
response = json.loads(content)
print_get_policy_response(response)
def print_get_policy_response(response: str) -> None:
pp = pprint.PrettyPrinter(indent=4)
pp.pprint(response)
def main(argv: Sequence[str]):
if len(argv) > 3:
raise app.UsageError('Too many command-line arguments.')
dc = create_delegated_credentials(_SA_FILE.value, _ADMIN_EMAIL.value)
dc.refresh(google.auth.transport.requests.Request())
call_get_policy_api(dc.token, _POLICY_NAME.value)
if __name__ == '__main__':
app.run(main)
할당량
각 Google Cloud 프로젝트에 대해 Cloud Identity Policy API는 초당 1개의 쿼리 (QPS)를 지원합니다. Cloud Identity Policy API는 고객이 여러 개의 Google Cloud 프로젝트를 만들더라도 고객별로 총 1개의 QPS를 지원합니다.
할당량 증가는 지원되지 않습니다.