huaweicloud-sdk-python-v3/huaweicloud-sdk-cts/huaweicloudsdkcts/v3/cts_client.py

745 lines
26 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# coding: utf-8
from __future__ import absolute_import
import datetime
import re
import importlib
import six
from huaweicloudsdkcore.client import Client, ClientBuilder
from huaweicloudsdkcore.exceptions import exceptions
from huaweicloudsdkcore.utils import http_utils
from huaweicloudsdkcore.sdk_stream_request import SdkStreamRequest
class CtsClient(Client):
"""
:param configuration: .Configuration object for this client
:param pool_threads: The number of threads to use for async requests
to the API. More threads means more concurrent API requests.
"""
PRIMITIVE_TYPES = (float, bool, bytes, six.text_type) + six.integer_types
NATIVE_TYPES_MAPPING = {
'int': int,
'long': int if six.PY3 else long,
'float': float,
'str': str,
'bool': bool,
'date': datetime.date,
'datetime': datetime.datetime,
'object': object,
}
def __init__(self):
super(CtsClient, self).__init__()
self.model_package = importlib.import_module("huaweicloudsdkcts.v3.model")
self.preset_headers = {'User-Agent': 'HuaweiCloud-SDK-Python'}
@classmethod
def new_builder(cls, clazz=None):
if clazz is None:
return ClientBuilder(cls)
if clazz.__name__ != "CtsClient":
raise TypeError("client type error, support client type is CtsClient")
return ClientBuilder(clazz)
def create_notification(self, request):
"""创建关键操作通知
配置关键操作通知可在发生特定操作时使用预先创建好的SMN主题向用户手机、邮箱发送消息也可直接发送http/https消息。常用于实时感知高危操作、触发特定操作或对接用户自有审计分析系统。
:param CreateNotificationRequest request
:return: CreateNotificationResponse
"""
return self.create_notification_with_http_info(request)
def create_notification_with_http_info(self, request):
"""创建关键操作通知
配置关键操作通知可在发生特定操作时使用预先创建好的SMN主题向用户手机、邮箱发送消息也可直接发送http/https消息。常用于实时感知高危操作、触发特定操作或对接用户自有审计分析系统。
:param CreateNotificationRequest request
:return: CreateNotificationResponse
"""
all_params = ['create_notification_request_body']
local_var_params = {}
for attr in request.attribute_map:
if hasattr(request, attr):
local_var_params[attr] = getattr(request, attr)
collection_formats = {}
path_params = {}
query_params = []
header_params = {}
form_params = {}
body_params = None
if 'body' in local_var_params:
body_params = local_var_params['body']
if isinstance(request, SdkStreamRequest):
body_params = request.get_file_stream()
response_headers = []
header_params['Content-Type'] = http_utils.select_header_content_type(
['application/json;charset=UTF-8'])
auth_settings = []
return self.call_api(
resource_path='/v3/{project_id}/notifications',
method='POST',
path_params=path_params,
query_params=query_params,
header_params=header_params,
body=body_params,
post_params=form_params,
response_type='CreateNotificationResponse',
response_headers=response_headers,
auth_settings=auth_settings,
collection_formats=collection_formats,
request_type=request.__class__.__name__)
def create_tracker(self, request):
"""创建追踪器
云审计服务开通后系统会自动创建一个追踪器用来关联系统记录的所有操作。目前一个云账户在一个Region下支持创建一个管理类追踪器和多个数据类追踪器。 云审计服务支持在管理控制台查询近7天内的操作记录。如需保存更长时间的操作记录您可以在创建追踪器之后通过对象存储服务Object Storage Service以下简称OBS将操作记录实时保存至OBS桶中。
:param CreateTrackerRequest request
:return: CreateTrackerResponse
"""
return self.create_tracker_with_http_info(request)
def create_tracker_with_http_info(self, request):
"""创建追踪器
云审计服务开通后系统会自动创建一个追踪器用来关联系统记录的所有操作。目前一个云账户在一个Region下支持创建一个管理类追踪器和多个数据类追踪器。 云审计服务支持在管理控制台查询近7天内的操作记录。如需保存更长时间的操作记录您可以在创建追踪器之后通过对象存储服务Object Storage Service以下简称OBS将操作记录实时保存至OBS桶中。
:param CreateTrackerRequest request
:return: CreateTrackerResponse
"""
all_params = ['create_tracker_request_body']
local_var_params = {}
for attr in request.attribute_map:
if hasattr(request, attr):
local_var_params[attr] = getattr(request, attr)
collection_formats = {}
path_params = {}
query_params = []
header_params = {}
form_params = {}
body_params = None
if 'body' in local_var_params:
body_params = local_var_params['body']
if isinstance(request, SdkStreamRequest):
body_params = request.get_file_stream()
response_headers = []
header_params['Content-Type'] = http_utils.select_header_content_type(
['application/json;charset=UTF-8'])
auth_settings = []
return self.call_api(
resource_path='/v3/{project_id}/tracker',
method='POST',
path_params=path_params,
query_params=query_params,
header_params=header_params,
body=body_params,
post_params=form_params,
response_type='CreateTrackerResponse',
response_headers=response_headers,
auth_settings=auth_settings,
collection_formats=collection_formats,
request_type=request.__class__.__name__)
def delete_notification(self, request):
"""删除关键操作通知
云审计服务支持删除已创建的关键操作通知。
:param DeleteNotificationRequest request
:return: DeleteNotificationResponse
"""
return self.delete_notification_with_http_info(request)
def delete_notification_with_http_info(self, request):
"""删除关键操作通知
云审计服务支持删除已创建的关键操作通知。
:param DeleteNotificationRequest request
:return: DeleteNotificationResponse
"""
all_params = ['notification_id']
local_var_params = {}
for attr in request.attribute_map:
if hasattr(request, attr):
local_var_params[attr] = getattr(request, attr)
collection_formats = {}
path_params = {}
query_params = []
if 'notification_id' in local_var_params:
query_params.append(('notification_id', local_var_params['notification_id']))
header_params = {}
form_params = {}
body_params = None
if isinstance(request, SdkStreamRequest):
body_params = request.get_file_stream()
response_headers = []
header_params['Content-Type'] = http_utils.select_header_content_type(
['application/json'])
auth_settings = []
return self.call_api(
resource_path='/v3/{project_id}/notifications',
method='DELETE',
path_params=path_params,
query_params=query_params,
header_params=header_params,
body=body_params,
post_params=form_params,
response_type='DeleteNotificationResponse',
response_headers=response_headers,
auth_settings=auth_settings,
collection_formats=collection_formats,
request_type=request.__class__.__name__)
def delete_tracker(self, request):
"""删除追踪器
云审计服务目前仅支持删除已创建的数据类追踪器。删除追踪器对已有的操作记录没有影响,当您重新开通云审计服务后,依旧可以查看已有的操作记录。
:param DeleteTrackerRequest request
:return: DeleteTrackerResponse
"""
return self.delete_tracker_with_http_info(request)
def delete_tracker_with_http_info(self, request):
"""删除追踪器
云审计服务目前仅支持删除已创建的数据类追踪器。删除追踪器对已有的操作记录没有影响,当您重新开通云审计服务后,依旧可以查看已有的操作记录。
:param DeleteTrackerRequest request
:return: DeleteTrackerResponse
"""
all_params = ['tracker_name', 'tracker_type']
local_var_params = {}
for attr in request.attribute_map:
if hasattr(request, attr):
local_var_params[attr] = getattr(request, attr)
collection_formats = {}
path_params = {}
query_params = []
if 'tracker_name' in local_var_params:
query_params.append(('tracker_name', local_var_params['tracker_name']))
if 'tracker_type' in local_var_params:
query_params.append(('tracker_type', local_var_params['tracker_type']))
header_params = {}
form_params = {}
body_params = None
if isinstance(request, SdkStreamRequest):
body_params = request.get_file_stream()
response_headers = []
header_params['Content-Type'] = http_utils.select_header_content_type(
['application/json'])
auth_settings = []
return self.call_api(
resource_path='/v3/{project_id}/trackers',
method='DELETE',
path_params=path_params,
query_params=query_params,
header_params=header_params,
body=body_params,
post_params=form_params,
response_type='DeleteTrackerResponse',
response_headers=response_headers,
auth_settings=auth_settings,
collection_formats=collection_formats,
request_type=request.__class__.__name__)
def list_notifications(self, request):
"""查询关键操作通知
查询创建的关键操作通知规则。
:param ListNotificationsRequest request
:return: ListNotificationsResponse
"""
return self.list_notifications_with_http_info(request)
def list_notifications_with_http_info(self, request):
"""查询关键操作通知
查询创建的关键操作通知规则。
:param ListNotificationsRequest request
:return: ListNotificationsResponse
"""
all_params = ['notification_type', 'notification_name']
local_var_params = {}
for attr in request.attribute_map:
if hasattr(request, attr):
local_var_params[attr] = getattr(request, attr)
collection_formats = {}
path_params = {}
if 'notification_type' in local_var_params:
path_params['notification_type'] = local_var_params['notification_type']
query_params = []
if 'notification_name' in local_var_params:
query_params.append(('notification_name', local_var_params['notification_name']))
header_params = {}
form_params = {}
body_params = None
if isinstance(request, SdkStreamRequest):
body_params = request.get_file_stream()
response_headers = []
header_params['Content-Type'] = http_utils.select_header_content_type(
['application/json'])
auth_settings = []
return self.call_api(
resource_path='/v3/{project_id}/notifications/{notification_type}',
method='GET',
path_params=path_params,
query_params=query_params,
header_params=header_params,
body=body_params,
post_params=form_params,
response_type='ListNotificationsResponse',
response_headers=response_headers,
auth_settings=auth_settings,
collection_formats=collection_formats,
request_type=request.__class__.__name__)
def list_quotas(self, request):
"""查询租户追踪器配额信息
查询租户追踪器配额信息。
:param ListQuotasRequest request
:return: ListQuotasResponse
"""
return self.list_quotas_with_http_info(request)
def list_quotas_with_http_info(self, request):
"""查询租户追踪器配额信息
查询租户追踪器配额信息。
:param ListQuotasRequest request
:return: ListQuotasResponse
"""
all_params = []
local_var_params = {}
for attr in request.attribute_map:
if hasattr(request, attr):
local_var_params[attr] = getattr(request, attr)
collection_formats = {}
path_params = {}
query_params = []
header_params = {}
form_params = {}
body_params = None
if isinstance(request, SdkStreamRequest):
body_params = request.get_file_stream()
response_headers = []
header_params['Content-Type'] = http_utils.select_header_content_type(
['application/json'])
auth_settings = []
return self.call_api(
resource_path='/v3/{project_id}/quotas',
method='GET',
path_params=path_params,
query_params=query_params,
header_params=header_params,
body=body_params,
post_params=form_params,
response_type='ListQuotasResponse',
response_headers=response_headers,
auth_settings=auth_settings,
collection_formats=collection_formats,
request_type=request.__class__.__name__)
def list_traces(self, request):
"""查询事件列表
通过事件列表查询接口可以查出系统记录的7天内资源操作记录。
:param ListTracesRequest request
:return: ListTracesResponse
"""
return self.list_traces_with_http_info(request)
def list_traces_with_http_info(self, request):
"""查询事件列表
通过事件列表查询接口可以查出系统记录的7天内资源操作记录。
:param ListTracesRequest request
:return: ListTracesResponse
"""
all_params = ['trace_type', 'limit', '_from', 'next', 'to', 'tracker_name', 'service_type', 'user', 'resource_id', 'resource_name', 'resource_type', 'trace_id', 'trace_name', 'trace_rating']
local_var_params = {}
for attr in request.attribute_map:
if hasattr(request, attr):
local_var_params[attr] = getattr(request, attr)
collection_formats = {}
path_params = {}
query_params = []
if 'trace_type' in local_var_params:
query_params.append(('trace_type', local_var_params['trace_type']))
if 'limit' in local_var_params:
query_params.append(('limit', local_var_params['limit']))
if '_from' in local_var_params:
query_params.append(('from', local_var_params['_from']))
if 'next' in local_var_params:
query_params.append(('next', local_var_params['next']))
if 'to' in local_var_params:
query_params.append(('to', local_var_params['to']))
if 'tracker_name' in local_var_params:
query_params.append(('tracker_name', local_var_params['tracker_name']))
if 'service_type' in local_var_params:
query_params.append(('service_type', local_var_params['service_type']))
if 'user' in local_var_params:
query_params.append(('user', local_var_params['user']))
if 'resource_id' in local_var_params:
query_params.append(('resource_id', local_var_params['resource_id']))
if 'resource_name' in local_var_params:
query_params.append(('resource_name', local_var_params['resource_name']))
if 'resource_type' in local_var_params:
query_params.append(('resource_type', local_var_params['resource_type']))
if 'trace_id' in local_var_params:
query_params.append(('trace_id', local_var_params['trace_id']))
if 'trace_name' in local_var_params:
query_params.append(('trace_name', local_var_params['trace_name']))
if 'trace_rating' in local_var_params:
query_params.append(('trace_rating', local_var_params['trace_rating']))
header_params = {}
form_params = {}
body_params = None
if isinstance(request, SdkStreamRequest):
body_params = request.get_file_stream()
response_headers = []
header_params['Content-Type'] = http_utils.select_header_content_type(
['application/json'])
auth_settings = []
return self.call_api(
resource_path='/v3/{project_id}/traces',
method='GET',
path_params=path_params,
query_params=query_params,
header_params=header_params,
body=body_params,
post_params=form_params,
response_type='ListTracesResponse',
response_headers=response_headers,
auth_settings=auth_settings,
collection_formats=collection_formats,
request_type=request.__class__.__name__)
def list_trackers(self, request):
"""查询追踪器
开通云审计服务成功后您可以在追踪器信息页面查看追踪器的详细信息。详细信息主要包括追踪器名称用于存储操作事件的OBS桶名称和OBS桶中的事件文件前缀。
:param ListTrackersRequest request
:return: ListTrackersResponse
"""
return self.list_trackers_with_http_info(request)
def list_trackers_with_http_info(self, request):
"""查询追踪器
开通云审计服务成功后您可以在追踪器信息页面查看追踪器的详细信息。详细信息主要包括追踪器名称用于存储操作事件的OBS桶名称和OBS桶中的事件文件前缀。
:param ListTrackersRequest request
:return: ListTrackersResponse
"""
all_params = ['tracker_name', 'tracker_type']
local_var_params = {}
for attr in request.attribute_map:
if hasattr(request, attr):
local_var_params[attr] = getattr(request, attr)
collection_formats = {}
path_params = {}
query_params = []
if 'tracker_name' in local_var_params:
query_params.append(('tracker_name', local_var_params['tracker_name']))
if 'tracker_type' in local_var_params:
query_params.append(('tracker_type', local_var_params['tracker_type']))
header_params = {}
form_params = {}
body_params = None
if isinstance(request, SdkStreamRequest):
body_params = request.get_file_stream()
response_headers = []
header_params['Content-Type'] = http_utils.select_header_content_type(
['application/json'])
auth_settings = []
return self.call_api(
resource_path='/v3/{project_id}/trackers',
method='GET',
path_params=path_params,
query_params=query_params,
header_params=header_params,
body=body_params,
post_params=form_params,
response_type='ListTrackersResponse',
response_headers=response_headers,
auth_settings=auth_settings,
collection_formats=collection_formats,
request_type=request.__class__.__name__)
def update_notification(self, request):
"""修改关键操作通知
云审计服务支持修改已创建关键操作通知配置项通过notification_id的字段匹配修改对象notification_id必须已经存在。
:param UpdateNotificationRequest request
:return: UpdateNotificationResponse
"""
return self.update_notification_with_http_info(request)
def update_notification_with_http_info(self, request):
"""修改关键操作通知
云审计服务支持修改已创建关键操作通知配置项通过notification_id的字段匹配修改对象notification_id必须已经存在。
:param UpdateNotificationRequest request
:return: UpdateNotificationResponse
"""
all_params = ['update_notification_request_body']
local_var_params = {}
for attr in request.attribute_map:
if hasattr(request, attr):
local_var_params[attr] = getattr(request, attr)
collection_formats = {}
path_params = {}
query_params = []
header_params = {}
form_params = {}
body_params = None
if 'body' in local_var_params:
body_params = local_var_params['body']
if isinstance(request, SdkStreamRequest):
body_params = request.get_file_stream()
response_headers = []
header_params['Content-Type'] = http_utils.select_header_content_type(
['application/json;charset=UTF-8'])
auth_settings = []
return self.call_api(
resource_path='/v3/{project_id}/notifications',
method='PUT',
path_params=path_params,
query_params=query_params,
header_params=header_params,
body=body_params,
post_params=form_params,
response_type='UpdateNotificationResponse',
response_headers=response_headers,
auth_settings=auth_settings,
collection_formats=collection_formats,
request_type=request.__class__.__name__)
def update_tracker(self, request):
"""修改追踪器
云审计服务支持修改已创建追踪器的配置项包括OBS桶转储、关键事件通知、事件转储加密、通过LTS对管理类事件进行检索、事件文件完整性校验以及追踪器启停状态等相关参数修改追踪器对已有的操作记录没有影响。修改追踪器完成后系统立即以新的规则开始记录操作。
:param UpdateTrackerRequest request
:return: UpdateTrackerResponse
"""
return self.update_tracker_with_http_info(request)
def update_tracker_with_http_info(self, request):
"""修改追踪器
云审计服务支持修改已创建追踪器的配置项包括OBS桶转储、关键事件通知、事件转储加密、通过LTS对管理类事件进行检索、事件文件完整性校验以及追踪器启停状态等相关参数修改追踪器对已有的操作记录没有影响。修改追踪器完成后系统立即以新的规则开始记录操作。
:param UpdateTrackerRequest request
:return: UpdateTrackerResponse
"""
all_params = ['update_tracker_request_body']
local_var_params = {}
for attr in request.attribute_map:
if hasattr(request, attr):
local_var_params[attr] = getattr(request, attr)
collection_formats = {}
path_params = {}
query_params = []
header_params = {}
form_params = {}
body_params = None
if 'body' in local_var_params:
body_params = local_var_params['body']
if isinstance(request, SdkStreamRequest):
body_params = request.get_file_stream()
response_headers = []
header_params['Content-Type'] = http_utils.select_header_content_type(
['application/json;charset=UTF-8'])
auth_settings = []
return self.call_api(
resource_path='/v3/{project_id}/tracker',
method='PUT',
path_params=path_params,
query_params=query_params,
header_params=header_params,
body=body_params,
post_params=form_params,
response_type='UpdateTrackerResponse',
response_headers=response_headers,
auth_settings=auth_settings,
collection_formats=collection_formats,
request_type=request.__class__.__name__)
def call_api(self, resource_path, method, path_params=None, query_params=None, header_params=None, body=None,
post_params=None, response_type=None, response_headers=None, auth_settings=None,
collection_formats=None, request_type=None):
"""Makes the HTTP request and returns deserialized data.
:param resource_path: Path to method endpoint.
:param method: Method to call.
:param path_params: Path parameters in the url.
:param query_params: Query parameters in the url.
:param header_params: Header parameters to be placed in the request header.
:param body: Request body.
:param post_params dict: Request post form parameters,
for `application/x-www-form-urlencoded`, `multipart/form-data`.
:param auth_settings list: Auth Settings names for the request.
:param response_type: Response data type.
:param response_headers: Header should be added to response data.
:param collection_formats: dict of collection formats for path, query,
header, and post parameters.
:param request_type: Request data type.
:return:
Return the response directly.
"""
return self.do_http_request(
method=method,
resource_path=resource_path,
path_params=path_params,
query_params=query_params,
header_params=header_params,
body=body,
post_params=post_params,
response_type=response_type,
response_headers=response_headers,
collection_formats=collection_formats,
request_type=request_type)