[toc]
注意: debug仅在调试阶段开启, 生产环境请关闭(iam_logger.setLevel(logging.ERROR));
生产环境开启带来问题:
- 日志量过大
- 影响请求速度(性能大幅降低)
- 敏感信息泄漏(权限相关)
from iam import IAM, Request, Subject, Action, Resource
import sys
import logging
iam_logger = logging.getLogger("iam")
iam_logger.setLevel(logging.DEBUG)
debug_handler = logging.StreamHandler(sys.stdout)
debug_handler.setFormatter(logging.Formatter('%(levelname)s [%(asctime)s] [IAM] %(message)s'))
iam_logger.addHandler(debug_handler)在调用sdk进行鉴权时, 会在终端打印debug日志
DEBUG [2020-05-21 14:23:22,833] [IAM] calling IAM.is_allowed(request)......
DEBUG [2020-05-21 14:23:22,833] [IAM] the request: {'action': {'id': 'access_developer_center'}, 'environment': {}, 'system': 'bk_paas', 'resources': [], 'subject': {'type': 'user', 'id': u'admin'}}
INFO [2020-05-21 14:23:22,918] [IAM] request: [method=`POST`, url=`http://{IAM_HOST}:8080/api/v1/policy/query`, data=`{'action': {'id': 'access_developer_center'}, 'environment': {}, 'system': 'bk_paas', 'resources': [], 'subject': {'type': 'user', 'id': u'admin'}}`]response: [status_code=`200`, request_id=`8e0c2a6f599248ecaad7579488956927`, content=`{"code":0,"message":"ok","data":{"field":"","op":"any","value":[]},"debug":{"time":"2020-05-21T14:23`]
DEBUG [2020-05-21 14:23:22,919] [IAM] the request id: `8e0c2a6f599248ecaad7579488956927`
DEBUG [2020-05-21 14:23:22,919] [IAM] the curl: `curl -X POST -H 'X-BK-APP-CODE: {APP_CODE}' -H 'X-BK-APP-SECRET: {APP_SECRET}' -H 'X-Bk-IAM-Version: 1' -d '{"action": {"id": "access_developer_center"}, "environment": {}, "system": "bk_paas", "resources": [], "subject": {"type": "user", "id": "admin"}}' 'http://127.0.0.1:8080/api/v1/policy/query?debug=true'`
DEBUG [2020-05-21 14:23:22,919] [IAM] do http request: method=`http_post`, url=`http://{IAM_HOST}/api/v1/policy/query`, data=`{"action": {"id": "access_developer_center"}, "environment": {}, "system": "bk_paas", "resources": [], "subject": {"type": "user", "id": "admin"}}`
DEBUG [2020-05-21 14:23:22,920] [IAM] http request result: ok=`True`, _data=`{"message": "ok", "code": 0, "data": {"field": "", "value": [], "op": "any"}}`
DEBUG [2020-05-21 14:23:22,920] [IAM] http request took 86 ms
DEBUG [2020-05-21 14:23:22,920] [IAM] the return policies: {u'field': u'', u'value': [], u'op': u'any'}
DEBUG [2020-05-21 14:23:22,920] [IAM] the return expr: ( any [])
DEBUG [2020-05-21 14:23:22,921] [IAM] the return expr render: (None any [])
DEBUG [2020-05-21 14:23:22,921] [IAM] the return expr eval: True
DEBUG [2020-05-21 14:23:22,921] [IAM] the return expr eval took 0 ms
其中:
request_id可以在问题排查时, 用于发给IAM服务端获取相关服务端处理日志the curl可以直接复制, 用于复现执行, 注意这里暴露了系统的访问鉴权信息(所以生产环境禁止开启debug, 仅供开发联调使用)http request took是单个请求的耗时, 可以用于确认慢的问题the return expr包含最终表达式的原始数据/渲染数据/求值结果/求值耗时
注意: 仅用于开发/联调环境排查定位问题, 一定不要在生产环境中开启, 将会导致API性能急剧下降
可以设置环境变量:
IAM_API_DEBUG=trueorBKAPP_IAM_API_DEBUG=true: 会在api url中加入?debug=true, 使得请求返回json中多返回debug字段, 包含策略相关的完整上下文信息, 用于精确定位api执行过程IAM_API_FORCE=trueorBKAPP_IAM_API_FORCE=true: 会在api url中加入?force=true, 请求将不走缓存, 直接从db中获取数据, 用于排查缓存类bug
生产环境, 可能需要保留IAM流水日志, 用于后续问题排查及定位
此时, 可以将iam logger配置到对应框架的日志配置中
注意:
- 日志级别设置为INFO: 请求流水日志, 包含成功/失败等(建议)
- 日志级别设置为ERROR: 报错日志, 网络请求及非200返回
以Django为例
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
......
},
'handlers': {
......
# 增加iam heandler, 落地日志到日志目录, 文件名 xxx_iam.log, 并且按固定大小rotate, 保留固定个数
'iam': {
'class': LOG_CLASS,
'formatter': 'verbose',
'filename': os.path.join(LOGGING_DIR, 'xxx_iam.log'),
'maxBytes': LOG_MAX_BYTES,
'backupCount': LOG_BACKUP_COUNT
},
},
'loggers': {
......
# 配置iam logger, 使用iam handler, 日志级别同项目配置的LOGGER_LEVEL
'iam': {
'handlers': ['iam'],
'level': LOGGER_LEVEL,
'propagate': True,
},
}
}查询是否有某个操作权限(没有资源实例), 例如访问开发者中心
from iam import IAM, Request, Subject, Action, Resource
class Permission(object):
def __init__(self):
self._iam = IAM(APP_CODE, APP_SECRET, BK_IAM_HOST, BK_PAAS_HOST)
def _make_request_without_resources(self, username, action_id):
request = Request(
SYSTEM_ID,
Subject("user", username),
Action(action_id),
None,
None,
)
return request
def allowed_access_developer_center(self, username):
"""
访问开发者中心权限
"""
request = self._make_request_without_resources(username, "access_developer_center")
return self._iam.is_allowed(request)查询是否有某个资源的某个操作权限(有资源实例), 例如管理某个应用
class Permission(object):
def __init__(self):
self._iam = IAM(APP_CODE, APP_SECRET, BK_IAM_HOST, BK_PAAS_HOST)
def _make_request_with_resources(self, username, action_id, resources):
request = Request(
SYSTEM_ID,
Subject("user", username),
Action(action_id),
resources,
None,
)
return request
def allowed_develop_app(self, username, app_code):
"""
app开发权限
"""
r = Resource(SYSTEM_ID, 'app', app_code, {})
resources = [r]
request = self._make_request_with_resources(username, "develop_app", resources)
return self._iam.is_allowed(request)对一批资源同时进行鉴权
可以调用batch_is_allowed (注意这个封装不支持跨系统资源依赖, 只支持接入系统自己的本地资源
subject = Subject("user", "tom")
action = Action("flow_edit")
resource1 = Resource("bk_sops", "flow", "1", {})
resource2 = Resource("bk_sops", "flow", "2", {})
resource3 = Resource("bk_sops", "flow", "3", {})
# 注意这里resources字段空
request = Request(
"bk_sops",
subject,
action,
[],
None
)
iam = IAM("bk_sops", "app_secret", "bk_iam_host", "bk_paas_host")
iam.batch_is_allowed(request, [[resource1], [resource2], [resource3]]))
# {'1': True, '2': False, '3': True}对一个资源的多个操作同时进行鉴权
可以调用resource_multi_actions_allowed进行批量操作权限的鉴权
subject = Subject("user", "tom")
action1 = Action("flow_edit")
action2 = Action("flow_view")
action3 = Action("flow_delete")
resource1 = Resource("bk_sops", "flow", "1", {})
request = MultiActionRequest(
"bk_sops",
subject,
[action1, action2, action3],
[resource1],
None
)
iam = IAM(
"bk_sops", "app_secret",
"bk_iam_host", "bk_paas_host"
)
iam.resource_multi_actions_allowed(request)
# {'flow_edit': True, 'flow_view': True, 'flow_delete': False}对于批量资源的多个操作同时进行鉴权, 例如进入资源列表也，可能需要在前端展示当前用户关于列表中的资源的一批操作的权限信息
可以调用batch_resource_multi_actions_allowed
subject = Subject("user", "tom")
action1 = Action("flow_edit")
action2 = Action("flow_view")
action3 = Action("flow_delete")
resource1 = Resource("bk_sops", "flow", "1", {})
resource2 = Resource("bk_sops", "flow", "2", {})
resource3 = Resource("bk_sops", "flow", "3", {})
request = MultiActionRequest(
"bk_sops",
subject,
[action1, action2, action3],
[],
None
)
iam = IAM(
"bk_sops", "app_secret",
"bk_iam_host", "bk_paas_host"
)
iam.batch_resource_multi_actions_allowed(request, [[resource1], [resource2], [resource3]]))
# {'1': {'flow_edit': True, 'flow_view': True, 'flow_delete': False}, '2': {'flow_edit': True, 'flow_view': True, 'flow_delete': False}, '3': {'flow_edit': True, 'flow_view': True, 'flow_delete': False}}对于非敏感权限
可以调用is_allowed_with_cache(request), 默认缓存10s.
(注意: 不要用于新建关联权限的资源is_allowed判定, 否则可能新建一个资源新建关联生效之后跳转依旧没权限; 更多用于管理权限/未依赖资源的权限权限判断)
# cache 10s, 10s内不会重新请求权限并进行鉴权计算, 能加快前端的体验
self._iam.is_allowed_with_cache(request)查询某个用户有权限的资源列表, 例如查询某个用户有开发权限的应用列表
def app_list(self, username):
"""
用户有权限的应用列表
拉回策略, 自己算!
"""
request = self._make_request_without_resources(username, "develop_app")
# 注册的资源类型是app, 所以返回策略中是app.id eq xxx;
# 但是数据库中queryset过滤的字段名是code, 所以需要做个转换;
# 数据库中将父级资源的 id 记录再了 parent_id 字段上，也需要定义转换
key_mapping = {"app.id": "code", "app.path": "parent_id"}
# 需要会将权限中心保存的 /resource_type,id/ path 转换为数据库中存储的 id 形式
def path_value_hook(value):
# get id in "/project,id/"
return value[1:-1].split(",")[1]
value_hooks = {"app.path": path_value_hook}
# 默认make_filter使用的是DjangoQuerySetConverter
filters = self._iam.make_filter(request, key_mapping=key_mapping, value_hooks=value_hooks)
# 如果从服务端查不到策略, 代表没有任何权限
if not filters:
return []
# 直接用django queryset查
apps = App.objects.filter(filters).all()
print("apps", apps)
return [app.code for app in apps]没有权限时, 在前端展示要申请的权限列表, 需要访问IAM接口, 拿到申请权限url; 用户点击跳转到IAM SaaS对应页面申请权限
- 申请不带资源实例的权限
from iam.apply.models import ActionWithoutResources, ActionWithResources, Application, RelatedResourceType
from iam.apply.models import ResourceInstance, ResourceNode
class Permission(object):
def __init__(self):
self._iam = IAM(APP_CODE, APP_SECRET, BK_IAM_HOST, BK_PAAS_HOST)
def make_no_resource_application(self, action_id):
# 1. make application
action = ActionWithoutResources(action_id)
actions = [action]
application = Application(SYSTEM_ID, actions)
return application
def generate_apply_url(self, bk_token, application):
"""
处理无权限 - 跳转申请列表
"""
# 2. get url
ok, message, url = self._iam.get_apply_url(application, bk_token)
if not ok:
logger.error("iam generate apply url fail: %s", message)
return IAM_APP_URL
return url- 申请带资源实例的权限
from iam.apply.models import ActionWithoutResources, ActionWithResources, Application, RelatedResourceType
from iam.apply.models import ResourceInstance, ResourceNode
class Permission(object):
def __init__(self):
self._iam = IAM(APP_CODE, APP_SECRET, BK_IAM_HOST, BK_PAAS_HOST)
def make_resource_application(self, action_id, resource_type, resource_id, resource_name):
# 1. make application
# 这里支持带层级的资源, 例如 biz: 1/set: 2/host: 3
# 如果不带层级, list中只有对应资源实例
instance = ResourceInstance([ResourceNode(resource_type, resource_id, resource_name)])
# 同一个资源类型可以包含多个资源
related_resource_type = RelatedResourceType(SYSTEM_ID, resource_type, [instance])
action = ActionWithResources(action_id, [related_resource_type])
actions = [action, ]
application = Application(SYSTEM_ID, actions)
return application
def generate_apply_url(self, bk_token, application):
"""
处理无权限 - 跳转申请列表
"""
# 2. get url
ok, message, url = self._iam.get_apply_url(application, bk_token)
if not ok:
logger.error("iam generate apply url fail: %s", message)
return IAM_APP_URL
return url使用 iam.utils.gen_perms_apply_data 可以生成生成无权限描述协议数据，其接收参数如下：
system：系统 IDsubject：Subject对象action_to_resources_list：无权动作与相关资源实例的映射列表，格式为：单个 action 中对应的 resources_list 必须是同类型的 Resource[ { "action": Action, "resources_list": [[resource1, resource2], [resource1, resource2]] }, ... ]
使用示例：
from iam.utils import gen_perms_apply_data
system = "test_system"
subject = Subject("user", "admin")
action1 = Action("action1")
action2 = Action("action2")
action3 = Action("action3")
resource1 = Resource("test_system", "r1", "r1id", {"name": "r1n"})
resource2 = Resource("test_system", "r2", "r2id", None)
resource3 = Resource("test_system", "r3", "r3id", {})
resource4 = Resource("another_system", "r4", "r4id", {"name": "r4n"})
resource5 = Resource("another_system", "r4", "r5id", {"name": "r5n"})
data = gen_perms_apply_data(
system,
subject,
[
{"action": action1, "resources_list": [[resource1, resource2, resource3, resource4]]}, # 含有拓扑层级的资源
{"action": action2, "resources_list": [[]]}, # 不关联资源类型的操作
{
"action": action3, # 带有跨系统依赖的资源
"resources_list": [
[resource1, resource3, resource4],
],
},
],
)sdk提供AuthFailedExceptionMiddleware 特殊处理鉴权不通过异常AuthFailedBaseException的时候，默认返回HTTP状态码为499，API的请求则可以通过在settings中配置参数BK_IAM_API_PREFIX进行前缀匹配，匹配通过时返回的状态码为200
使用示例：
# settings.py 配置如下：
BK_IAM_API_PREFIX = SITE_URL + 'openapi'- 将
iam.contrib.iam_migration加入INSTALLED_APPS中 - 在项目根目录的
support-files/iam/中添加 iam migration json 文件 - 执行
python manage.py iam_makemigrations {migration_json_file_name}- 其中
{migration_json_file_name}为新加入的 iam migration json 文件名 - 该命令会在
iam/contrib/iam_migration/migrations目录下生成用于执行向权限中心注册系统、资源和操作的 migration 文件，当应用第一次部署时，这些 migration 文件会随之执行。 - 注意：如果你的 iam sdk 不是以源码的方式嵌入项目中而是以 pip 的方式安装的，那么请额外配置
BK_IAM_MIGRATION_APP_NAME来设置用于存储 migration 文件的 APP
- 其中
基本配置
APP_CODE/SECRET_KEY应用在蓝鲸开发者中心申请应用的app_code/app_secretBK_IAM_SYSTEM_ID接入系统注册到权限中心使用的系统 ID(system_id)- 调用方式(二选一, 如果有 APIGateway, 优先使用 APIGateway):
- 直连:
BK_IAM_INNER_HOST权限中心后台的地址 - APIGateway:
BK_IAM_USE_APIGATEWAY = TrueBK_IAM_APIGATEWAY_URL = "http://bk-iam.{APIGATEWAY_DOMAIN}/{env}"
- 直连:
Migration 相关
BK_IAM_MIGRATION_JSON_PATH：如果你不想将 iam migration json 放置在support-files/iam/目录下，请在 Django Setting 中将该变量配置为你想要存放 iam migration json 文件的相对目录BK_IAM_MIGRATION_APP_NAME：如果你是以 pip 的方式安装 iam sdk，那么请单独新建一个 Django app，将BK_IAM_MIGRATION_APP_NAME设置为该 app 的 label，并将该 app 加入INSTALLED_APPS中，iam migrator 会将 Django migration 文件置于该 app 的migrations目录下。BK_IAM_RESOURCE_API_HOST：如果你无法确定 upsert_system 操作 data 中的provider_config.host的值，那么可以在 Django Setting 中配置这个变量，IAM Migration 会在执行 upsert_system 操作前将provider_config.host设置为BK_IAM_RESOURCE_API_HOSTBK_IAM_SKIP: 是否跳过iam migration, 某些版本(<1.1.15)sdk强依赖, 可以设置成False或None
其他
- TIPS：
- 可以使用
python manage.py startapp {app_name}命令来新建 django app - 如果 app 是已存在的，请确保该 app 目录下存在
migrations/__init__.py文件）
- 可以使用
SDK 提供了 Resource API Framework，它能够帮助你处理 Resource API 必须处理的一些公共逻辑（鉴权，参数校验，响应标准化等），还提供了提供不同 Web 框架（Django）API 的 Dispatcher。
Resource API Framework 中有两个核心概念：
- ResourceProvider：封装不同类型的资源在处理来自 IAM 不同类型的方法时的处理逻辑
- Dispatcher：针对不同的 Web 框架，提供 API 的定义
用户自定义的 Provider 必须继承自 iam.resource.provider.ResourceProvider，并且实现以下方法：
list_attr(**options)：处理来自 IAM 的 list_attr 请求
- 参数：
options：language(str)：国际化语言
- 返回值：返回
iam.resource.provider.ListResult的实例，其中results应满足 IAM list_attr 响应协议
list_attr_value(filter, page, **options)：处理来自 IAM 的 list_attr_value 请求
- 参数：
filter：过滤器对象filter.attr(str)：需要查询的资源属性idfilter.keyword(str)：资源属性值的搜索关键字filter.ids(list[string,int,bool])：资源属性值ID列表
page：分页对象page.limit(int)：查询数量page.offset(int)：查询偏移
options：language(str)：国际化语言
- 返回值：返回
iam.resource.provider.ListResult的实例，其中results应满足 IAM list_attr_value 响应协议
list_instance(filter, page, **options)：处理来自 IAM 的 list_instance 请求
- 参数：
filter：过滤器对象filter.parent(dict)：资源的直接上级，具体包含type和id，type为直接上级资源的类型，id为直接上级资源实例IDfilter.search(str)：资源实例的搜索关键字filter.resource_type_chain(list[dict])：配置search参数一起使用，resource_type_chain指定返回对象返回的祖先层级拓扑
page：分页对象page.limit(int)：查询数量page.offset(int)：查询偏移
options：language：国际化语言
- 返回值：返回
iam.resource.provider.ListResult的实例，其中results应满足 IAM list_instance 响应协议
fetch_instance_info(filter, page, **options)：处理来自 IAM 的 fetch_instance_info 请求
- 参数：
filter：过滤器对象filter.ids(list[str])：需要查询的资源实例的唯一标识列表filter.attrs(lsit[str])：需要查询的资源属性列表，比如["path", "os"]，空列表或无该参数则表示查询所有属性
options：language：国际化语言
- 返回值：返回
iam.resource.provider.ListResult的实例，其中results应满足 IAM fetch_instance_info 响应协议
list_instance_by_policy(filter, page, **options)：处理来自 IAM 的 list_instance_by_policy 请求
- 参数：
filter：过滤器对象filter.expression(dict)：资源的表达式，协议请查看
page：分页对象page.limit(int)：查询数量page.offset(int)：查询偏移
options：language：国际化语言
- 返回值：返回
iam.resource.provider.ListResult的实例，其中results应满足 IAM list_instance_by_policy 响应协议
search_instance(filter, page, **options)：处理来自 IAM 的 search_instance 请求
- 参数：
filter：过滤器对象filter.keyword(str)：资源实例的搜索关键字filter.parent(dict)：资源的直接上级，具体包含type和id，type为直接上级资源的类型，id为直接上级资源实例ID
page：分页对象page.limit(int)：查询数量page.offset(int)：查询偏移
- 返回值：返回
iam.resource.provider.ListResult的实例，其中results应满足 IAM search_instance 响应协议
fetch_instance_list(filter, page, **options)：处理来自 IAM 的 fetch_instance_list 请求，在审计中心生成静态资源快照时，需要实现此方法
- 参数：
filter：过滤器对象filter.start_time(int)：资源实例变更时间的开始时间（包含start_time）filter.end_time(int)：资源实例变更时间的结束时间（包含end_time）
page：分页对象page.limit(int)：查询数量page.offset(int)：查询偏移
- 返回值：返回
iam.resource.provider.ListResult的实例，其中results应满足 IAM fetch_instance_list 响应协议
fetch_resource_type_schema(**options)：处理来自 IAM 的 fetch_resource_type_schema 请求，在审计中心显示静态资源时，需要实现此方法
- 返回值：返回
iam.resource.provider.SchemaResult的实例，输出结果可以通过 JSON Schema Validator 校验 注意：为满足审计中心需求，字段描述新增description_en、code两个 Key，description_en指字段英文描述，code（可选）指该字段为代码内容，在审计中心将按代码格式显示- DEMO（Response.data）
{
"type": "object",
"properties": {
"id": {
"type": "string",
"description": "ID",
"description_en": "ID",
},
"script": {
"type": "string",
"description": "脚本",
"description_en": "Script",
"code": "shell"
}
}
}
除此之外，如果 Provider 中定义了 pre_{method} 方法（method 可选值（list_attr, list_attr_value, list_instance, fetch_instance_info, list_instance_by_policy），Dispatcher 会在调用对应的 {method} 方法前调用其对应的 pre 方法进行预处理，下面的例子检测 list_instance 中传入的 page 对象，如果 limit 过大，则拒绝该请求：
class TaskResourceProvider(ResourceProvider):
def pre_list_instance(self, filter, page, **options):
if page.limit == 0 or page.limit > 50:
raise InvalidPageException("limit in page too large")
...下面是一种资源类型的 Provider 定义示例：
from iam.resource.provider import ResourceProvider, ListResult, SchemaResult
from task.models import Tasks
class TaskResourceProvider(ResourceProvider):
def list_instance(self, filter, page, **options):
"""配置权限时的下拉列表
注意, 有翻页; 需要返回count
"""
queryset = Tasks.objects.all()
count = queryset.count()
results = [
{"id": str(task.id), "display_name": task.name} for task in queryset[page.slice_from : page.slice_to]
]
return ListResult(results=results, count=count)
def search_instance(self, filter, page, **options):
"""配置权限时的搜索框
注意, 有翻页; 需要返回count
"""
queryset = Tasks.objects.filter(name__contains=filter.keyword).all()
count = queryset.count()
results = [
{"id": str(task.id), "display_name": task.name} for task in queryset[page.slice_from : page.slice_to]
]
return ListResult(results=results, count=count)
def fetch_instance_info(self, filter, **options):
"""申请权限时, 回调这个接口进行资源信息正确性/合法性校验
无需count, 但是目前ListResult初始化这个参数暂时是必选的(FIXME)
"""
ids = []
if filter.ids:
ids = [int(i) for i in filter.ids]
results = [{"id": str(task.id), "display_name": task.task_name} for task in Tasks.objects.filter(id__in=ids)]
return ListResult(results=results, count=0)
def list_attr(self, **options):
"""通过属性配置权限会用到, 没有属性权限管控不需要实现
属性列表
"""
return ListResult(results=[], count=0)
def list_attr_value(self, filter, page, **options):
"""通过属性配置权限会用到, 没有属性权限管控不需要实现
属性值列表
注意, 有翻页; 需要返回count
"""
return ListResult(results=[], count=0)
def list_instance_by_policy(self, filter, page, **options):
"""权限预览, 暂时没有用到, 可以不实现
注意, 有翻页; 需要返回count
"""
return ListResult(results=[], count=0)
def fetch_instance_list(self, filter, page, **options):
"""根据过滤条件搜索实例
注意, 有翻页; 需要返回count
"""
return ListResult(results=[], count=0)
def fetch_resource_type_schema(self, **options):
"""获取资源类型 schema 定义
schema定义
"""
return SchemaResult(properties={})iam.contrib.django.dispatcher.DjangoBasicResourceApiDispatcher 是适用于 Django 的 Resource API Dispatcher
iam：对应系统的iam.IAM实例system：对应系统的 ID (注意,这里是systemID, 不是app_code, 一定不能传app_code, 某些环境systemID不一定是app_code)
注册类型为 resource_type 的 Provider
resource_type：资源类型provider：iam.resource.provider.ResourceProvider的子类实例
返回能够作为 Resource API 的 view 函数
decorators：需要装饰返回 view 函数的装饰器列表
from blueapps.account.decorators import login_exempt
from iam import IAM
from iam.contrib.django.dispatcher import DjangoBasicResourceApiDispatcher
from iam.resource.provider import ResourceProvider, ListResult, SchemaResult
from iam.contrib.converter.queryset import PathEqDjangoQuerySetConverter
from django.conf.urls import url, include
iam = IAM(
"my_system", "app_secret",
"iam_api_host", "paas_host"
)
def flow_path_value_hook(value):
# get id in "/project,id/"
return value[1:-1].split(",")[1]
class FlowResourceProvider(ResourceProvider):
def list_attr(self, **options):
"""
flow 资源没有属性，返回空
"""
return ListResult(results=[], count=0)
def list_attr_value(self, filter, page, **options):
"""
flow 资源没有属性，返回空
"""
return ListResult(results=[], count=0)
def list_instance(self, filter, page, **options):
"""
flow 上层资源为 project
"""
queryset = []
with_path = False
if not (filter.parent or filter.search or filter.resource_type_chain):
queryset = TaskTemplate.objects.all()
elif filter.parent:
parent_id = filter.parent["id"]
if parent_id:
queryset = TaskTemplate.objects.filter(project_id=str(parent_id))
elif filter.search and filter.resource_type_chain:
# 返回结果需要带上资源拓扑路径信息
with_path = True
# 过滤 project flow 名称
project_keywords = filter.search.get("project", [])
flow_keywords = filter.search.get("flow", [])
project_filter = Q()
flow_filter = Q()
for keyword in project_keywords:
project_filter |= Q(name__icontains=keyword)
for keyword in flow_keywords:
flow_filter |= Q(pipeline_template__name__icontains=keyword)
project_ids = Project.objects.filter(project_filter).values_list("id", flat=True)
queryset = TaskTemplate.objects.filter(project_id__in=list(project_ids)).filter(flow_filter)
results = [
{"id": str(flow.id), "display_name": flow.name} for flow in queryset[page.slice_from : page.slice_to]
]
if with_path:
results = [
{
"id": str(flow.id),
"display_name": flow.name,
"path": [[{"type": "project", "id": str(flow.project_id), "display_name": flow.project.name}]],
}
for flow in queryset[page.slice_from : page.slice_to]
]
return ListResult(results=results, count=len(results))
def fetch_instance_info(self, filter, **options):
"""
flow 没有定义属性，只处理 filter 中的 ids 字段
"""
ids = []
if filter.ids:
ids = [int(i) for i in filter.ids]
results = [{"id": str(flow.id), "display_name": flow.name} for flow in TaskTemplate.objects.filter(id__in=ids)]
return ListResult(results=results, count=0)
def list_instance_by_policy(self, filter, page, **options):
"""
flow
"""
expression = filter.expression
if not expression:
return ListResult(results=[])
key_mapping = {
"flow.id": "id",
"flow.owner": "pipeline_template__creator",
"flow.path": "project__id",
}
# 这里使用 PathEqDjangoQuerySetConverter 是为了将对 flow.path 的 starts_with 操作符转换为 eq 操作符
converter = PathEqDjangoQuerySetConverter(key_mapping, {"flow.path": flow_path_value_hook})
filters = converter.convert(expression)
results = [
{"id": str(flow.id), "display_name": flow.name}
for flow in TaskTemplate.objects.filter(filters)[page.slice_from : page.slice_to]
]
return ListResult(results=results, count=len(results))
def search_instance(self, filter, page, **options):
# TODO
return ListResult(results=[], count=0)
def fetch_instance_list(self, filter, page, **options):
"""
flow
"""
queryset = TaskTemplate.objects.filter(updated_at__gte=filter.start_time).filter(updated_at__lte=filter.end_time)
results = []
for flow in queryset[page.slice_from : page.slice_to]:
results.append(
{
"id": str(flow.id),
"display_name": flow.name,
"creator": flow.creator,
"created_at": flow.created_at,
"updater": flow.updater,
"updated_at": flow.updated_at,
"data": flow.to_json()
}
)
return ListResult(results=results, count=queryset.count())
def fetch_resource_type_schema(self, **options):
properties = {
"id": {
"type": "string",
"description": "ID",
"description_en": "ID",
},
"script": {
"type": "string",
"description": "脚本",
"description_en": "Script",
"code": "shell"
}
}
return SchemaResult(properties=properties)
dispatcher = DjangoBasicResourceApiDispatcher(iam, "my_system")
dispatcher.register("flow", FlowResourceProvider())
urlpatterns = [
url(r'^resource/api/v1/$', dispatcher.as_view([login_exempt]))
]未来 APIGateway 高性能网关会作为一个基础的蓝鲸服务, 权限中心将会把后台及 SaaS 的所有开放 API 接入到网关中bk-iam
此时, 对于接入方, 不管是鉴权/申请权限还是其他接口, 都可以通过同一个网关访问到.
理解成本更低, 且相关的调用日志/文档/流控/监控等都可以在 APIGateway 统一管控.
网关地址类似: http://bk-iam.{APIGATEWAY_DOMAIN}/{env}, 其中 env值 prod(生产)/stage(预发布)
SDK 目前做了兼容, 可以修改初始化 SDK 的参数, 切换流量到 APIGateway
# 测试环境, 使用stage
IAM(app_code, app_secret, bk_apigateway_url="http://bk-iam.{APIGATEWAY_DOMAIN}/stage"):
# 正式环境, 使用prod
IAM(app_code, app_secret, bk_apigateway_url="http://bk-iam.{APIGATEWAY_DOMAIN}/prod"):如果使用了sdk提供的 IAM Migration, 需要设置环境变量
BK_IAM_USE_APIGATEWAY = TrueBK_IAM_APIGATEWAY_URL = "http://bk-iam.{APIGATEWAY_DOMAIN}/{env}"
当前SDK默认使用 v2 鉴权 api, 如果开发者环境的权限中心后台版本小于 v1.2.6, 则不支持直接使用v2 api, 需要配置api_version指定使用v1 api
IAM(APP_CODE, APP_SECRET, BK_IAM_HOST, BK_PAAS_HOST, api_version="v1")