资源政策评估库
项目描述
资源政策评估库
rpe-lib 由Policy Engines (rpe.engines.Engine)、Resources (rpe.resources.Resource)和组成Extractors (rpe.resources.Extractor)。
Resources生成有关给定类型资源的当前状态的详细信息。资源几乎可以是任何东西，但该项目的最初目标是支持 Google Cloud Platform (GCP) 资源。该实现是有意通用的，以允许支持其他资源类型
Policy EnginesResources根据他们配置的策略进行评估。给定一个资源，他们可以返回一个列表，其中列出了Evaluations (rpe.policy.Evaluation)适用于该资源的每个策略的名称（按资源类型），以及该资源是否合规。Resource还定义了一个remediate()函数，它应该能够以字典描述如何操作资源以使其兼容。
Extractors解析结构化数据并返回资源列表，以及可选的有关数据源的元数据。
例如，对于 GCP 资源，该remediate()函数需要有关在 Google REST API 中为给定资源调用什么方法以及要传递什么参数的详细信息。因此，对于 Google Cloud Storage 存储桶，强制存储桶版本控制的策略可以将修复定义为patch()使用适当的参数调用 buckets 方法以启用版本控制。
政策引擎
Policy Enginesrpe-lib目前包含 2个：
开放策略代理 (opa) 引擎
Open Policy Agent 引擎使用 Open Policy Agent REST API 来评估给定资源的策略。为了使用这个引擎，您需要使用 rpe-lib 基本策略以及您想要评估和可选执行的任何策略来运行 opa 服务器。opa 引擎将结果Resource.get()作为输入传递给 opa 服务器。所有策略都需要在rpe.policy命名空间中定义，包括一些特定的规则，并对input文档进行操作：
- apply_to：策略适用的资源类型列表
- description : 一个人类可读的策略描述
- compliant
input：如果文档中定义的资源符合策略，则返回 true - exclude ：如果有理由从评估中排除资源，则返回 true，对于 GCP，我们定义了可以将资源标记为已排除的资源标签
- remediate (optional) : 返回一个 JSON 对象，解释如何为给定策略修复资源
示例策略：
# This is a real policy included with rpe-lib though slightly simplified
package rpe.policy.storage_buckets_require_object_versioning
description = "Require object versioning for storage buckets"
applies_to = ["storage.googleapis.com/Bucket"]
default compliant = false
default excluded = false
compliant {
input.resource.versioning.enabled = true
}
excluded {
input.resource.labels["forseti-enforcer"] = "disabled"
}
remediate = {
"_remediation_spec": "v2",
"steps": [
"method": "patch",
"params": {
"bucket": input.resource.name,
"body": {"versioning": {"enabled": true}},
}
]
}
Python 引擎
OPA 引擎非常强大，但受限于 OPA 的能力。对于更复杂的用例，您可能需要使用 python 引擎。该引擎需要一个 python 包的路径，该包包含执行评估和可选修复的类。
如前所述，python 策略实际上是类。这是一个示例 python 策略类：
# Stubbed out example of the above OPA engine policy as a python engine policy
class GCPBucketVersioningPolicy:
description = 'Require object versioning for storage buckets'
applies_to = ['cloudresourcemanager.googleapis.com/Project']
@classmethod
def compliant(cls, resource):
return resource.get()['resource']['versioning']['enabled'] == True
@classmethod
def excluded(cls, resource):
return resource.get()['resource']['labels']['forseti-enforcer'] == 'disabled'
@classmethod
def remediate(cls, resource):
# execute cloudfuntion to enable versioning,
# or shell execute gsutil,
# or anything you can do with python
pass
当你配置你的 RPE 对象时，你提供了你的 python 包的路径。该软件包不需要安装，它将动态加载。您提供的路径应该包含一个__init__.py文件，其中包含您希望使用的每个策略类。这些可以直接在该文件中定义或导入。由于 RPElib 负责导入包，因此您将能够在 python 文件中使用相对导入。例如，您的目录结构可能如下所示：
/etc/rpe/policies-python
* __init__.py
* policy1.py
* policy2.py
你__init__.py可能看起来像这样：
# __init__.py
from .policy1 import MyFirstPythonPolicy
from .policy2 import AnotherPythonPolicy, YetAnotherPolicy
资源
Resources必须定义以下函数
get()：返回描述资源当前状态的元数据remediate(remediation_spec)：应用补救（规范特定于资源类型/实现）type()：返回资源的类型，策略引擎使用它来确定哪些策略适用于给定资源
例子
使用 OPA 引擎
这假设您opa的路径中有二进制文件
# First, start opa with our policies
opa run --server ./policy/
现在我们需要创建一个 RPE 实例，并将 opa 引擎配置为使用本地 OPA 服务器：
from rpe import RPE
from rpe.resources import GoogleAPIResource
config = {
'policy_engines': [
{
'type': 'opa',
'url': 'http://localhost:8181/v1/data'
}
]
}
rpe = RPE(config)
# Create a resource object for the resource we want to evaluate
res = GoogleAPIResource.from_cai_data(
'//storage.googleapis.com/my-test-bucket',
'storage.googleapis.com/Bucket',
project_id='my-test-project',
)
evals = rpe.evaluate(res)
for e in evals:
print(f'Policy: {e.policy_id}, Compliant: {e.compliant}')
if not e.compliant and e.remediable:
e.remediate()
使用 Python 引擎
使用 Python 策略引擎与上面类似：
from rpe import RPE
from rpe.resources import GoogleAPIResource
config = {
'policy_engines': [
{
'type': 'python',
'path': '/etc/rpe/policies-python'
}
]
}
rpe = RPE(config)
# Create resource objects, and evaluate as needed
您的政策可能如下所示：
/etc/rpe/policies-python/ init .py
from .gcs_policy import GCPBucketVersioningPolicy
/etc/rpe/policies-python/gcs_policy.py
class GCPBucketVersioningPolicy:
description = 'Require object versioning for storage buckets'
applies_to = ['cloudresourcemanager.googleapis.com/Project']
@classmethod
def compliant(cls, resource):
# Return true/false if the resource is compliant
@classmethod
def excluded(cls, resource):
# Return true/false if the resource is excluded
@classmethod
def remediate(cls, resource):
# Enable versioning
使用 rpe-lib 的应用程序
- Forseti 实时执行器- Forseti 实时执行器使用 rpe-lib 评估和执行 Google Cloud 资源的策略。它使用 Stackdriver 日志导出到 Pub/Sub 主题来触发强制执行。
