Python 脚本，用于通过 REST API 更新 Google Cloud Platform 中的 Cloud Ar

Question

我想编写一个Python脚本，该脚本将使用Compute Engine REST API更新或添加新规则到Google Cloud Platform中的CloudArmor。

但我有几个疑问：

• 官方的 Google Cloud Client Library for Python 是否适合此目的(如果不是您所建议的)？

• 为此，我应该安装哪个 Python 包？

• 此标准身份验证(包含我的私钥并正确设置环境变量GOOGLE_APPLICATION_CREDENTIALS的 JSON 文件)是否足以连接到此 API 以实现我想要实现的目的？

我在问路。

Answer 1

Python 脚本将是一个很好的方法，您可以使用 Google API 的 Python 客户端库来实现这一点。

适用于 Python 的 Google Cloud Client Library 可能还不够，但 Google Cloud Client Libraries 是用于调用 Google Cloud API 的最新和推荐的客户端库。

计算引擎 v1securityPoliciesREST API 资源提供了分别添加新规则、检索和更新现有规则所需的方法列表，例如addRule、getRule和patchRule。

查看计算引擎 API 的 PyDoc 参考，获取方法的完整列表，并说明如何使用它们。

此外，Python 包(Google API Python Client Library)是您需要使用的。

您提到的标准身份验证提供了一种有用的方法，用于通过 Python 客户端库授权对计算引擎 API 的请求。

根据官方文档：GCP 客户端库使用称为应用程序默认凭据 (ADC) 的策略来查找应用程序的凭据。当代码使用客户端库时，策略将按以下顺序检查凭据：

1. 首先，ADC检查环境变量是否 GOOGLE_APPLICATION_CREDENTIALS已设置。如果设置了变量，则 ADC 使用变量指向的服务帐户文件。下一个 部分介绍如何设置环境变量。
2. 如果未设置环境变量，ADC 将使用默认服务 计算引擎、Kubernetes Engine、App Engine 和 云函数为在这些上运行的应用程序提供 服务业。
3. 如果 ADC 无法使用上述任一凭据，则会发生错误。

最后，确保您选择使用的用户账户具有在计算引擎上配置 Cloud Armor 所需的正确 IAM 权限。

Answer 2

import googleapiclient.discovery
from google.oauth2 import service_account
from pprint import pprint
import logging
if LOCAL == 'True':
credentials = service_account.Credentials.from_service_account_file(
'path-to-service-account.json')
compute = googleapiclient.discovery.build('compute',
'v1',
credentials=credentials)
else:
compute = googleapiclient.discovery.build('compute',
'v1')

class cloud_armor():
def __init__(self, domain):
self.domain = domain
self.project_id = <PROJECT_ID>
self.policy_name = <POLICY_NAME>
def add_rule(self):
#Find minimum current priority
current_policy = cloud_armor.get_policy(self)
current_rules = current_policy['rules']
rule_priorities = []
for rule in current_rules:
rule_priorities.append(rule['priority'])
priority = int(min(rule_priorities)) - 1
body = {
"description": "{}".format(self.domain),
"priority": priority,
"match": {
"expr": {
"expression": "request.headers['referer']=="{}"".format(
self.domain)
}
},
"action": "allow",
"preview": False,
"kind": "compute#securityPolicy"
}
try:
policies = compute.securityPolicies()
rule = policies.addRule(project=self.project_id,
securityPolicy=self.policy_name,
body=body
).execute()
return rule
except Exception as err:
for i in range(0, len(err.args)):
logging.error(err.args[i])
pprint(err.args[i])
print("==Policy Rule Failed to Add==")
raise Exception("Policy Rule Failed to Add")
def get_policy(self):
try:
policies = compute.securityPolicies()
policy = policies.get(project=self.project_id,
securityPolicy=self.policy_name
).execute()
return policy
except Exception as err:
for i in range(err.args):
logging.error(err.args[i])
pprint(err.args[i])
print("==Failed to Fetch Policy==")
raise Exception("Failed to Fetch Policy")