Cloud One Workload Security の Python SDK のソースコードを読んでみた

らら子です。
[昨日の記事](https://qiita.com/rarako/items/fdc36b04aaf9010d7c93)に続きまして、本日はCloud One Workload Securityの自動化APIを試していきます。

セキュリティソフトの管理なんてわざわざAPIで自動化することあるの？という疑問は持ちつつ公式ドキュメントを読み進めたところ、以下のような作業がAPIで自動化できることがメリットとして挙げられていました。

- ポリシーの設定とコンピュータの保護設定
- 脆弱性の発見とパッチの適用
- 定期的なメンテナンスタスクの自動化

へー、という感じなのですが、CloudWatchなどのイベントをトリガーにEC2にインストールしたセキュリティエージェントの保護設定を動的に切り替えていく、みたいな運用ができたらかっこいいですね！（運用の安定性はともかく・・・）

というわけでさっそくAPIを使っていきましょうということで、Cloud One Workload SecurityのAPIについて、[公式ドキュメント：The API and SDK](https://cloudone.trendmicro.com/docs/workload-security/api-first-steps/)を読みました。

どうやらPythonのSDKが提供されているようなので、こちらを利用していこうと思います。
（私はPythonとJSしか書けません！）

## Python SDKをセットアップする

Python SDKの設定は、[Python SDK ドキュメント](https://cloudone.trendmicro.com/docs/jp/workload-security/sdk-python/)を参照しました。

動作要件が「Python 3.4 または 3.7のバージョン」となってます。
やや古いですね。

一応手元のPython 3.9の環境でも問題なく動いたのは確認したのですが、念のためDockerを使ってPython 3.7のランタイム環境を用意しました。

``` bash
docker pull python:3.7
```

次に、[Python SDK ドキュメント](https://cloudone.trendmicro.com/docs/jp/workload-security/sdk-python/)のページから、Python SDKの本体をダウンロードします。
PypyじゃなくてZIPで配信してるんですね。

適当なディレクトリでZIPファイルを展開します。

``` bash
wget https://cloudone.trendmicro.com/docs/downloads/sdk/ws/v1/c1ws-py-sdk.zip
unzip c1ws-py-sdk.zip
rm c1ws-py-sdk.zip
```

展開後のファイルツリーはこんな感じでした。

``` bash
$ tree
.
├── deepsecurity
│   ├── __init__.py
│   ├── api
│   │   ├── 中略
│   ├── api_client.py
│   ├── configuration.py
│   ├── models
│   │   ├── 中略
│   └── rest.py
└── setup.py
3 directories, 314 files
```

SDKを展開したディレクトリをボリュームとしてDockerコンテナを起動し、以降はこのコンテナの中で作業を進めていきます。

まずはコンテナにログインします。

``` bash
docker run --rm -it -v `pwd`:/root python:3.7 bash -c "cd /root && bash"
#root@a07271f34eba:~# ls
#deepsecurity/  setup.py
```

コンテナにログインができたら、setup.pyと同じディレクトリでSDKをpip installします。

インストールが完了したらpip freezeを叩いてみると、`deep-security-api`というモジュールと、依存関係のパッケージがインストールされたことが確認できます。

``` bash
python3 -m pip install .

# pip freeze
WARNING: The directory '/root/.cache/pip' or its parent directory is not owned or is not writable by the current user. The cache has been disabled. Check the permissions and owner of that directory. If executing pip with sudo, you should use sudo's -H flag.
certifi==2021.5.30
deep-security-api @ file:///root
python-dateutil==2.8.2
six==1.16.0
urllib3==1.26.6
```

これでSDKの準備は完了しました。
公式ドキュメントではここからPyCharmのセットアップ方法が記載されていますが、私はPyCharmは使っていないのでパスします。

## CloudOne で APIアクセスキーを作成する

APIを利用するには、当然アクセスキーが必要です。

アクセスキーは、AdministrationタブからAPI Keysを開き、作成することができます。
アクセスキーごとに権限を制限できるようですが、今回はFull Access権限を指定します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1820420/6998945e-9bb7-7d02-b81b-158f77520ed6.png)

アクセスキーと合わせて必要なのは、APIのアクセスエンドポイントのURLです。

エンドポイントのURLは全然見つからなくて、探すのに苦労しました。

現在は`https://cloudone.trendmicro.com/api`のURLがAPIのエンドポイントになっているようです。

## APIを実行してみる

これでAPIを実行する準備が整ったので、さっそく呼び出してみようと思います。

サンプルコードは[API reference](https://cloudone.trendmicro.com/docs/workload-security/api-reference/tag/Administrators)に書いてあったので、こちらを利用しました。
※ アクセスキーについては環境変数に格納しています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1820420/93ef4167-3f3d-8989-a7b7-fb67640d2de8.png)

実際に利用したコードはこちらです。

これは、管理者ユーザーの一覧を取得するためのAPIのようですね。

``` python
from __future__ import print_function
import os
import sys, warnings
import deepsecurity
from deepsecurity.rest import ApiException
from pprint import pprint

# Setup
if not sys.warnoptions:
	warnings.simplefilter("ignore")
configuration = deepsecurity.Configuration()
configuration.host = os.getenv('ENDPOINT_URL')

# Authentication
configuration.api_key['api-secret-key'] = os.getenv('APIKEY')

# Initialization
# Set Any Required Values
api_instance = deepsecurity.AdministratorsApi(deepsecurity.ApiClient(configuration))
api_version = 'v1'

try:
	api_response = api_instance.list_administrators(api_version)
	pprint(api_response)
except ApiException as e:
	print("An exception occurred when calling AdministratorsApi.list_administrators: %s\n" % e)
```

試しに実行してみたところ、こんな感じの出力結果が得られました！
（JSONっぽい形式ですが、よくみると標準のJSON形式ではなさそう。）

``` bash
$ python3 admin_test.py
{'administrators': [{'active': True,
                     'created': 1629031887394,
                     'description': '',
                     'directory_info': None,
                     'directory_name': None,
                     'email_address': '',
                     'external': True,
                     'external_user_id': '',
                     'full_name': '',
                     'id': 1,
                     'last_password_change': None,
                     'last_sign_in': 1629121495862,
                     'locale': 'en-US',
                     'mfa_type': 'none',
                     'mobile_number': '',
                     'pager_number': '',
                     'password': None,
                     'password_never_expires': True,
                     'phone_number': '',
                     'primary_contact': False,
                     'read_only': False,
                     'receive_notifications': False,
                     'report_pdf_password': None,
                     'report_pdf_password_enabled': False,
                     'role_id': 1,
                     'time_format': '24',
                     'time_zone': 'UTC',
                     'type': 'normal',
                     'unlock_time': None,
                     'unsuccessful_sign_in_attempts': 0,
                     'username': 'urn:cloudone:xxxxxxxxxxxx',
                     'utc_offset': 'UTC+0.00'}]}
```

`last_password_change`とか`mfa`とか`role_id`とかのパラメータが確認できるので、もしかしたら社内ポリシーに違反していないか、などをチェックするような運用を想定しているのかもしれませんね。

## PythonSDKの構造を見る

本当はこれからいろんなAPIを試してみることをやりたかったのですが、利用できるメソッドの数が結構多かったのと、正直飽きてしまったこともあり、ここからはPython SDKのソースコードを読んでいきたいと思います。

### deepsecurity.Configuration()

まずは、先ほどの`admin_test.py`を見てみます。

```python
# Setup
if not sys.warnoptions:
	warnings.simplefilter("ignore")
configuration = deepsecurity.Configuration()
configuration.host = os.getenv('ENDPOINT_URL')

# Authentication
configuration.api_key['api-secret-key'] = os.getenv('APIKEY')
```

`deepsecurity.Configuration()`で`<class 'deepsecurity.configuration.Configuration'>`というオブジェクトを生成して、`host`と`api_key`をセットしていることがわかります。

そこで、まずはこのクラスについて見ていくことにします。

ちょっと長いですが、本体は`configration.py`の`Configration`クラスでした。
`host`や`api_key`、`username`などのメンバ変数が確認できます。

コード量が多かったので割愛していますが、興味深い点としては、`to_debug_report()`などのデバッグ用と思われるメソッドも実装されている点ですね。
どうやって使うんだろ…。

```python
class Configuration(six.with_metaclass(TypeWithDefault, object)):
    """NOTE: This class is auto generated by the swagger code generator program.

    Ref: https://github.com/swagger-api/swagger-codegen
    Do not edit the class manually.
    """

    def __init__(self):
        """Constructor"""
        # Default Base url
        self.host = "https://app.deepsecurity.trendmicro.com/api"
        # Temp file folder for downloading files
        self.temp_folder_path = None

        # Authentication Settings
        # dict to store API key(s)
        self.api_key = {}
        # dict to store API prefix (e.g. Bearer)
        self.api_key_prefix = {}
        # Username for HTTP basic authentication
        self.username = ""
        # Password for HTTP basic authentication
        self.password = ""

        # Logging Settings
        self.logger = {}
        self.logger["package_logger"] = logging.getLogger("deepsecurity")
        self.logger["urllib3_logger"] = logging.getLogger("urllib3")
        # Log format
        self.logger_format = '%(asctime)s %(levelname)s %(message)s'
        # Log stream handler
        self.logger_stream_handler = None
        # Log file handler
        self.logger_file_handler = None
        # Debug file location
        self.logger_file = None
        # Debug switch
        self.debug = False

        # SSL/TLS verification
        # Set this to false to skip verifying SSL certificate when calling API
        # from https server.
        self.verify_ssl = False
        # Set this to customize the certificate file to verify the peer.
        self.ssl_ca_cert = None
        # client certificate file
        self.cert_file = None
        # client key file
        self.key_file = None
        # Set this to True/False to enable/disable SSL hostname verification.
        self.assert_hostname = None

        # urllib3 connection pool's maximum number of connections saved
        # per pool. urllib3 uses 1 connection as default value, but this is
        # not the best value when you are making a lot of possibly parallel
        # requests to the same host, which is often the case here.
        # cpu_count * 5 is used as default value to increase performance.
        self.connection_pool_maxsize = multiprocessing.cpu_count() * 5

        # Proxy URL
        self.proxy = None
        # Safe chars for path_param
        self.safe_chars_for_path_param = ''
```

### ApiClient オブジェクトを作成する

次に、先ほど作成した`configuration`オブジェクトを引数として、`ApiClient`オブジェクトを作成しています。
そして、その`ApiClient`オブジェクトを引数として、さらに`AdministratorsApi`オブジェクトを`api_instance`オブジェクトとして生成している流れです。

実質的に、この`api_instance`オブジェクトがSDKの本体部分といって差し支えなさそうです。

```python
# Initialization
# Set Any Required Values
api_instance = deepsecurity.AdministratorsApi(
    deepsecurity.ApiClient(configuration)
)
api_version = 'v1'
```

`ApiClient`オブジェクトクラスのソースコードを見てみましょう。
こちらも全部で700行くらいあるので、一部のみ貼り付けています。

コメントを見ると、Swagger Codegenで自動生成されたコードのようですね。
[Swagger](https://swagger.io/specification/)というのはめちゃべんりなツールで、単一のSwagger SpecというAPIの仕様を定義したファイルから、APIリファレンスやAPIスタブを自動生成できるものです。

Cloud One の[API reference](https://cloudone.trendmicro.com/docs/workload-security/api-reference/tag/Administrators#operation/listAdministrators)も、このSwagger SpecからAPIリファレンスを生成する[Swagger UI](https://swagger.io/swagger-ui/)で作成されたものをカスタマイズしたものっぽいですね。

余談はともかく、このクラス定義を見たときに、着目すべき点は`self.rest_client = rest.RESTClientObject(configuration)`のように見えますね。

このRESTClientObjectが、実際にAPIエンドポイントに対してのHTTPリクエストを送信し、レスポンスを処理しているようです。

ソースコードは`rest.py`を見てください。

```python
class ApiClient(object):
    """Generic API client for Swagger client library builds.

    Swagger generic API client. This client handles the client-
    server communication, and is invariant across implementations. Specifics of
    the methods and models for each application are generated from the Swagger
    templates.

    NOTE: This class is auto generated by the swagger code generator program.
    Ref: https://github.com/swagger-api/swagger-codegen
    Do not edit the class manually.

    :param configuration: .Configuration object for this client
    :param header_name: a header to pass when making calls to the API.
    :param header_value: a header value to pass when making calls to
        the API.
    :param cookie: a cookie to include in the header when making calls
        to the API
    """

    PRIMITIVE_TYPES = (float, bool, bytes, six.text_type) + six.integer_types
    NATIVE_TYPES_MAPPING = {
        'int': int,
        'long': int if six.PY3 else long,  # noqa: F821
        'float': float,
        'str': str,
        'bool': bool,
        'date': datetime.date,
        'datetime': datetime.datetime,
        'object': object,
    }

    def __init__(self, configuration=None, header_name=None, header_value=None,
                 cookie=None):
        if configuration is None:
            configuration = Configuration()
        self.configuration = configuration

        self.pool = ThreadPool()
        self.rest_client = rest.RESTClientObject(configuration)
        self.default_headers = {}
        if header_name is not None:
            self.default_headers[header_name] = header_value
        self.cookie = cookie
        # Set default User-Agent.
        self.user_agent = 'Swagger-Codegen/20.0.242/python'

    def __del__(self):
        self.pool.close()
        self.pool.join()
```

### AdministratorsApi オブジェクトを作成する

最後に、`AdministratorsApi`オブジェクトのソースコードを見てみます。

このクラスが今回利用したSDKの実質的な本体と言っていいかと思いますが、このようなクラスは、APIの種類の数だけ用意されているようです。

例えば、スケジュールタスクの設定に関するAPIを使う際は、`ScheduledTasksApi`オブジェクトを呼び出しますし、コンピュータの情報を扱う場合は`ComputersApi`オブジェクトを呼び出す仕様になっているようです。

さて、`AdministratorsApi`オブジェクトのソースコードですが、これも結構長くて、700行弱ありますので割愛します。

クラスのコードを読んでみると、それぞれのメンバ関数が、Cloud OneのAPIと1:1で紐づいていそうなことがわかります。
（もしかしたらAPIリファレンスを読むよりSDKのソースコードを読んだ方がAPIの使い方がわかりやすいかも…）

```python
class AdministratorsApi(object):
    """NOTE: This class is auto generated by the swagger code generator program.

    Do not edit the class manually.
    Ref: https://github.com/swagger-api/swagger-codegen
    """

    def __init__(self, api_client=None):
        if api_client is None:
            api_client = ApiClient()
        self.api_client = api_client

    def create_administrator(self, administrator, api_version, **kwargs):  # noqa: E501
        """Create an Administrator  # noqa: E501

        Create a new administrator.  # noqa: E501
        This method makes a synchronous HTTP request by default. To make an
        asynchronous HTTP request, please pass async_req=True
        >>> thread = api.create_administrator(administrator, api_version, async_req=True)
        >>> result = thread.get()

        :param async_req bool
        :param Administrator administrator: The settings of the new administrator. (required)
        :param str api_version: The version of the api being called. (required)
        :return: Administrator
                 If the method is called asynchronously,
                 returns the request thread.
        """
        kwargs['_return_http_data_only'] = True
        if kwargs.get('async_req'):
            return self.create_administrator_with_http_info(administrator, api_version, **kwargs)  # noqa: E501
        else:
            (data) = self.create_administrator_with_http_info(administrator, api_version, **kwargs)  # noqa: E501
            return data
```

## まとめ

ほんとはもっと色々書くことあったんですが、長くなってしまったのでこれで終わります。

ちなみに、ここでプリントしてるAPIのレスポンス結果なんですが、見ればわかる通り、実は`<class 'deepsecurity.models.administrator.Administrator'>`のオブジェクトであって、JSON形式じゃないんですよね。

``` python
api_response = api_instance.describe_administrator(1, api_version)
pprint(api_response)
```

SDKのソースコードを見てみると、`to_dict()`とか`to_str()`みたいな関数が実装されているので、`api_response.to_dict()`のようにしてあげれば、辞書型や文字列型に変換できるようになっているようです。

``` python
def to_dict(self):
        """Returns the model properties as a dict"""
        result = {}

        for attr, _ in six.iteritems(self.swagger_types):
            value = getattr(self, attr)
            if isinstance(value, list):
                result[attr] = list(map(
                    lambda x: x.to_dict() if hasattr(x, "to_dict") else x,
                    value
                ))
            elif hasattr(value, "to_dict"):
                result[attr] = value.to_dict()
            elif isinstance(value, dict):
                result[attr] = dict(map(
                    lambda item: (item[0], item[1].to_dict())
                    if hasattr(item[1], "to_dict") else item,
                    value.items()
                ))
            else:
                result[attr] = value
        if issubclass(Administrator, dict):
            for key, value in self.items():
                result[key] = value

        return result

    def to_str(self):
        """Returns the string representation of the model"""
        return pprint.pformat(self.to_dict())
```

また、APIの仕様的には値が存在しない項目は本来は何も返ってこない仕様のようですがSDK側で次のように`swagger_types`や`attribute_map`が定義されており、APIのレスポンス内で欠損している値は勝手に`NONE`として埋めているようです。

``` python
swagger_types = {
    'username': 'str',
    'password': 'str',
    'full_name': 'str',
    'description': 'str',
    'role_id': 'int',
    'locale': 'str',
    'time_zone': 'str',
    'time_format': 'str',
    'password_never_expires': 'bool',
    'active': 'bool',
    'mfa_type': 'str',
    'phone_number': 'str',
    'mobile_number': 'str',
    'pager_number': 'str',
    'email_address': 'str',
    'primary_contact': 'bool',
    'receive_notifications': 'bool',
    'report_pdf_password_enabled': 'bool',
    'report_pdf_password': 'str',
    'created': 'int',
    'last_password_change': 'int',
    'last_sign_in': 'int',
    'unlock_time': 'int',
    'unsuccessful_sign_in_attempts': 'int',
    'directory_name': 'str',
    'directory_info': 'str',
    'external': 'bool',
    'external_user_id': 'str',
    'type': 'str',
    'read_only': 'bool',
    'id': 'int',
    'utc_offset': 'str'
}
```

つまり、SDKはただ単にAPIのリクエスト/レスポンスを取得しているわけではなく、裏でいろいろと補完しているみたいですね。

なんか癖があって使いにくく感じますが、Swaggerを使って作成したAPIスタブとかは大体こんな感じになるんでしょうか。

ともあれ、SDKのソースコードを読んでみていろいろと勉強になったので良かったです。