Alibaba SDKを使用して、パブリックTCP/IPアドレスを持つセキュリティグループを自動的に更新する

このチュートリアルでは、Windowsタスクスケジューラを使用して、ECSセキュリティグループを自動的にパブリックTCP/IPアドレスで最新の状態に保つための定期的なタスクをセットアップする方法を学びます。

*本ブログは英語版からの翻訳です。オリジナルは[こちら](https://www.alibabacloud.com/blog/automating-security-groups-updates-on-alibaba-cloud_593792)からご確認いただけます。一部機械翻訳を使用しております。翻訳の間違いがありましたら、ご指摘いただけると幸いです。*

[Alibaba Cloud Tech Share](https://www.alibabacloud.com/campaign/techshare?spm=a2c65.11461447.0.0.104124c9d6l5Te)執筆者、ジョン・ハンリー。[Tech Share](https://www.alibabacloud.com/campaign/techshare?spm=a2c65.11461447.0.0.104124c9d6l5Te)は、技術的な知識やベストプラクティスをクラウドコミュニティ内で共有することを奨励するAlibaba Cloudのインセンティブプログラムです。

[Alibaba Cloud Elastic Compute Service](https://www.alibabacloud.com/product/ecs?spm=a2c65.11461447.0.0.104124c9d6l5Te)（ECS）インスタンスを作成する際には、[セキュリティグループ](https://www.alibabacloud.com/help/doc-detail/58746.htm?spm=a2c65.11461447.0.0.104124c9d6l5Te)を作成または指定します。このセキュリティグループは、ECSインスタンスへのアクセスを制御するファイアウォールとして機能します。Linuxインスタンスの場合、ルールの1つはSSH(TCPポート22)アクセスを許可します。ベストプラクティスでは、自分が管理する TCP/IP アドレスからのみ SSH アクセスを許可することが必要です。セキュリティグループ（ファイアウォール）を介してTCP / IPアドレスのみを許可することにより、ECSインスタンスの公開フットプリントを削減します。

SSH用のセキュリティグループルールの作成は、Alibaba Cloud Console上で非常に簡単に行えます。しかし、そのルールを現在のTCP/IPアドレスで最新の状態に保つのは大変な作業です。まず、自分のパブリックTCP/IPアドレスを把握し、[Alibaba Cloud Console](https://home-intl.console.aliyun.com/?spm=a2c65.11461447.0.0.104124c9d6l5Te)にログインしてセキュリティグループを見つけ、パブリックIPアドレス用の新しいルールでセキュリティグループを修正し、最後に古いルールを削除する必要があります。

Alibaba Cloudには、セキュリティグループ内のルールをプログラムで作成、変更、削除するための[API](https://www.alibabacloud.com/help/doc-detail/25484.htm?spm=a2c65.11461447.0.0.104124c9d6l5Te)と[SDK](https://www.alibabacloud.com/help/doc-detail/25699.htm?spm=a2c65.11461447.0.0.104124c9d6l5Te)が用意されています。この記事では、Alibaba SDKを使用して、セキュリティグループをパブリックTCP/IPアドレスで自動的に更新する方法をデモします。その後、コマンドプロンプトから手動でプログラムを実行したり、タスクスケジューラを介して自動的に実行したりすることができます。この記事では、Windowsタスクスケジューラを使用して、セキュリティグループをパブリックTCP/IPアドレスで常に最新の状態に保つための定期的なタスクを設定する方法を紹介します。

#このプログラムはどのように機能するのか？
このプログラムは、現在のTCP/IPアドレスを "last_ip.txt "というファイルに保存します。次にプログラムを実行すると、現在の TCP/IP アドレスが前回と同じかどうかをチェックします。もし「true」であれば、セキュリティグループに変更は加えられません。アドレスが異なる場合は、新しいルールが作成され、前回のアドレスのルールが削除されます。これにより、古いエントリがセキュリティグループを汚染することなく、セキュリティグループを最新の状態に保ちます。

このプログラムはWindows ECSインスタンスにも対応しています。ソースコードのポート番号を変更するだけで、リモートデスクトップ(RDP)をサポートします。

あなたのパブリックTCP/IPアドレスは、NeoPrimeのウェブサーバーにアクセスしてURL `http://www.neoprime.io/test/getmyip.php` にアクセスすることで決定されます。 getmyip.phpのソースコードはダウンロードに含まれています。このURLは、ページにアクセスしたときに、あなたのパブリックTCP/IPアドレスを返すだけです。パブリックTCP/IPアドレスをHTMLマークアップなしのシンプルなテキストとして返すパブリックサーバーを使用することができます。

#ダウンロード
このリンクをクリックして、このプログラムのソースコードをダウンロードしてください。ソースコード (Zip - 3 KB)

最終更新日：2018年6月28日

必要な要件：Python 3.6以降（Python 2には対応していません）

プラットフォーム：Windows 10 でテスト済み

注意: Pythonのソースコードを含むZIPファイルなので、ウイルス対策ソフトはこのダウンロードに警告をするでしょう。


#リソースアクセス管理
このプログラムでは、セキュリティグループを変更するための権限が必要になります。セキュリティのベストプラクティスでは、必要最低限の権限のみを提供することを推奨しています。その勧告に従ってみましょう。このプログラムは、セキュリティグループのルールを記述する機能(DescribeSecurityGroupAttribute)、セキュリティグループのルールを作成する機能(AuthorizeSecurityGroup)、セキュリティグループのルールを削除する機能(RevokeSecurityGroup)、およびセキュリティグループをリストアップする機能(DescribeSecurityGroups)を必要とします。

以下のポリシーでは、必要なパーミッションをJSONで記述しています。この記事の後半では、カスタムポリシーを作成する際にこのJSONを使用します。

このリンクをクリックして、JSONポリシーファイルをダウンロードしてください。policy.jsonのダウンロード



```
{
  "Version": "1",
  "Statement": [
    {
      "Action": [
        "ecs:AuthorizeSecurityGroup",
        "ecs:DescribeSecurityGroups",
        "ecs:DescribeSecurityGroupAttribute",
        "ecs:RevokeSecurityGroup"
      ],
      "Resource": "*",
      "Effect": "Allow"
    }
  ]
}
```

#セキュリティポリシーの強化
セキュリティグループをより細かく管理したいと思うかもしれません。例えば、DevOps チームに 5 人の従業員がいて、それぞれが異なるサーバやサービスのセットを担当しているとします。各ユーザのリソースに対して異なるセキュリティグループを作成し、リソースレベルの権限を割り当てて、誰がどのセキュリティグループを変更できるかを制御することができます。次のポリシーでは、どのセキュリティグループを変更できるかを指定します。次に、異なるユーザーに割り当てられた異なるポリシーを作成します。これで、User-AがUser-Bのセキュリティグループを誤って変更することはできなくなりました。

```
{
  "Version": "1",
  "Statement": [
    {
      "Action": [
        "ecs:AuthorizeSecurityGroup",
        "ecs:DescribeSecurityGroups",
        "ecs:DescribeSecurityGroupAttribute",
        "ecs:RevokeSecurityGroup"
      ],
      "Resource": "acs:ecs:*:*:securitygroup/sg-rj01234567890abcdefg",
      "Effect": "Allow"
    }
  ]
}
```

#カスタムポリシーの作成
このパートでは、Alibaba Cloud Consoleを使って、セキュリティグループの管理に必要な権限だけを持つカスタムポリシーを作成していきます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/647921/e95333f0-98a8-edec-63c8-149b26b26a3e.png)

カスタムポリシーを作成します。

- [Alibaba Resource Access Management (RAM)コンソール](https://ram.console.aliyun.com/?spm=a2c65.11461447.0.0.104124c98W6kaX)にアクセスします。
- 「ポリシー」をクリックします。
- 「カスタムポリシー」タブをクリックします。
- 青色の「認証ポリシーの作成」ボタンをクリックします。
- 「ブランクテンプレート」をクリックします。
- 認可ポリシー名を入力：ManageSecurityGroupRules
- 説明を入力：セキュリティグループルールの管理
- ポリシーコンテンツを上記のJSONに置き換えます。
- 「認証ポリシーの作成」をクリックします。

#ユーザーの作成
このパートでは、Alibaba Consoleを使用して新しいユーザーを作成し、このユーザーにカスタムポリシーを割り当てます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/647921/243047d1-48eb-7d11-78a4-dc12b12d2382.png)

- [Alibaba Resource Access Management (RAM)コンソール](https://ram.console.aliyun.com/?spm=a2c65.11461447.0.0.104124c98W6kaX)に移動します。
- 「ユーザー」をクリックします。
- 青い「ユーザー作成」ボタンをクリックします。
- ユーザー名を入力してください: sg_auth
- 表示名を入力する: sg_auth
- 説明を入力：セキュリティグループのルールを管理するための sg_auth.py プログラムの権限
- ラジオボタン「このユーザーのアクセスキーを自動的に生成する」をクリックします。
- アクセスキー情報を保存します。これは後で必要になります。
- これでコンソールにユーザーの一覧が表示されました。先ほど作成したユーザーを配置します。「Authorize」をクリックします。
- 検索ダイアログで、先ほど作成したポリシーの最初の数文字を入力します："ManageSec"
- ManageSecurityGroupRulesをクリックします。
- 右矢印をクリックして、ポリシーを選択した列に移動します。
- OK をクリックします。


#ユーザー資格情報プロファイルの作成
このパートでは、Alibaba Cloud CLIを使用して、ユーザーのアクセスキーとシークレットキーで新しいプロファイルを作成します。また、デフォルトのリージョンと出力形式を指定します。

```
c:\Python27\Scripts\aliyuncli configure --profile sg_auth
Aliyun Access Key ID [None]: 
Aliyun Access Key Secret [None]: 
Default Region Id [None]: 
Default output format [None]: json
```

#プログラムの実行
pythonのサンプルプログラムを実行するには、コマンドプロンプトを開き、以下のように実行します。

```
python sg_auth.py --profile sg_auth
```

注意: pythonのソースコードを修正して、デフォルトで使用するプロファイル名を指定することができます。

```
PROFILE_NAME = 'default'    # specify the credentials profile name to use
```

#プログラムソースコード

```
############################################################
# Version 0.90
# Date Created: 2018-06-28
# Last Update:  2018-06-28
# https://www.neoprime.io
# Copyright (c) 2018, NeoPrime, LLC
# Author: John Hanley
############################################################

""" Add my public IP to an Alibaba Cloud Security Group """

import    os
import    json
import    logging
import    optparse
import    requests
from aliyunsdkcore.client import AcsClient
from aliyunsdkecs.request.v20140526 import AuthorizeSecurityGroupRequest
from aliyunsdkecs.request.v20140526 import RevokeSecurityGroupRequest
from aliyunsdkecs.request.v20140526 import DescribeSecurityGroupAttributeRequest

g_debug = False
g_print = True    # Set to diplay messages to console

LAST_IP_FILENAME = 'last_ip.txt'
IP_ENDPOINT = 'http://www.neoprime.io/test/getmyip.php'
PROFILE_NAME = 'default'    # specify the credentials profile name to use

sg_auth_params = {
    'sg_id': 'sg-rj01234567890abcdefg',    # Change this value to your security group
    'ip_protocol': 'tcp',
    #'port_range': '3389/3389',        # Use this port range for Remote Desktop (RDP)
    'port_range': '22/22',            # Use this port range for SSH
    'description': 'My public IP address',
    'source_cidr_ip': ''
}

logger = logging.getLogger('sg_auth')

def setup_logging():
    """ This creates sets the logging configuration """
    f1 = '%(asctime)s %(name)s %(levelname)s %(message)s'
    f2 = '%(message)s'

    if g_debug is False:
        if g_print is False:
            logging.basicConfig(filename='sg_auth.log', level=logging.INFO, format=f1)
        else:
            logging.basicConfig(level=logging.INFO, format=f2)
    else:
        if g_print is False:
            logging.basicConfig(filename='sg_auth.log', level=logging.DEBUG, format=f1)
        else:
            logging.basicConfig(level=logging.DEBUG, format=f2)

    logger.info('########################################')
    logger.info('Program start')

def usage():
    """ Command Usage Help """
    print("Usage: sg_auth [-d, --debug] [-p, --purge]")

def process_cmdline():
    """ Process the Command Line """
    parser = optparse.OptionParser()

    parser.set_defaults(debug=False, profile=False, region_id=False, slb_id=False)

    parser.add_option(
            '-d',
            '--debug',
            action='store_true',
            dest='debug',
            default=False,
            help="enable debugging")

    parser.add_option(
            '-p',
            '--purge',
            action='store_true',
            dest='purge',
            default=False,
            help="purge security groups rules with same port range")

    parser.add_option(
            '--profile',
            action = 'store',
            dest = 'profile',
            default = PROFILE_NAME,
            help = "specify the credentials profile name")

    (cmd_options, cmd_args) = parser.parse_args()

    return (cmd_options, cmd_args)

def get_current_ip():
    """ Get my public IP address """
    resp = requests.get(IP_ENDPOINT)
    resp.raise_for_status()
    ip = resp.content.strip().decode('utf-8')
    ip += '/32'
    return ip

def get_last_ip():
    """ Get my last public IP address that was saved """
    if not os.path.exists(LAST_IP_FILENAME):
        return None

    try:
        with open(LAST_IP_FILENAME, 'r') as fp:
            ip = fp.readline().strip()
    except:
        ip = None
    return ip

def save_new_ip(ip):
    """ Save my public IP address """
    with open(LAST_IP_FILENAME, 'w') as fp:
        fp.write(ip + '\n')

def sg_authorize(client, params):
    """ Authorize an IP address """
    # Initialize a request and set parameters
    request = AuthorizeSecurityGroupRequest.AuthorizeSecurityGroupRequest()

    request.set_SecurityGroupId(params['sg_id'])
    request.set_IpProtocol(params['ip_protocol'])
    request.set_PortRange(params['port_range'])
    request.set_Description(params['description'])
    request.set_SourceCidrIp(params['source_cidr_ip'])

    response = client.do_action_with_exception(request)

    if g_debug:
        logger.debug(response)

    r = json.loads(response)

    logger.debug('Request ID: %s', r['RequestId'])

def sg_revoke(client, params):
    """ Revoke an IP address """

    logger.info('Removing IP: %s', sg_auth_params['source_cidr_ip'])

    # Initialize a request and set parameters
    request = RevokeSecurityGroupRequest.RevokeSecurityGroupRequest()

    request.set_SecurityGroupId(params['sg_id'])
    request.set_IpProtocol(params['ip_protocol'])
    request.set_PortRange(params['port_range'])
    request.set_Description(params['description'])
    request.set_SourceCidrIp(params['source_cidr_ip'])

    response = client.do_action_with_exception(request)

    if g_debug:
        logger.debug(response)

    r = json.loads(response)

    logger.debug('Request ID: %s', r['RequestId'])

def purge_rules(client, params):
    """ Remove all rules with the same port range """

    logger.info('Purging all security groups rules for port range %s', params['port_range'])

    # Initialize a request and set parameters
    request = DescribeSecurityGroupAttributeRequest.DescribeSecurityGroupAttributeRequest()

    request.set_SecurityGroupId(params['sg_id'])

    response = client.do_action_with_exception(request)

    if g_debug:
        logger.debug(response)

    r = json.loads(response)

    logger.debug('Request ID: %s', r['RequestId'])

    for permission in r['Permissions']['Permission']:
        if permission['PortRange'] == params['port_range']:
            sg_auth_params['source_cidr_ip'] = permission['SourceCidrIp']
            sg_revoke(client, sg_auth_params)

def main_cmdline(options):
    """ This is the main function """

    last_ip = None
    current_ip = get_current_ip()

    if options.purge is False:
        last_ip = get_last_ip()
        if last_ip == current_ip:
            logger.info('Last ip and current ip are the same. No changes are required.')
            return 0

    # My library for processing Alibaba Cloud Services (ACS) credentials
    # This library is only used when running from the desktop and not from the cloud
    import mycred_acs

    # Load the Alibaba Cloud Credentials (AccessKey)
    logger.info('Loading credentials for profile %s', options.profile)
    credentials = mycred_acs.LoadCredentials(options.profile)

    if options.debug:
        logger.info('Access Key ID: %s', credentials['accessKeyId'])

    if credentials is False:
        logger.error('Error: Cannot load credentials')
        return 1

    # Initialize AcsClient instance
    client = AcsClient(
        credentials['accessKeyId'],
        credentials['accessKeySecret'],
        credentials['region'])

    if options.purge:
        purge_rules(client, sg_auth_params)

    sg_auth_params['source_cidr_ip'] = current_ip

    logger.info('Adding IP:   %s', sg_auth_params['source_cidr_ip'])

    sg_authorize(client, sg_auth_params)

    save_new_ip(current_ip)

    if last_ip is not None:
        sg_auth_params['source_cidr_ip'] = last_ip
        sg_revoke(client, sg_auth_params)

    return 0

# Setup logging
setup_logging()

# Process the command line
(g_options, g_args) = process_cmdline()

if g_options.debug:
    g_debug = True

ret = main_cmdline(g_options)
```

*アリババクラウドは日本に２つのデータセンターを有し、世界で60を超えるアベラビリティーゾーンを有するアジア太平洋地域No.1(2019ガートナー)のクラウドインフラ事業者です。
アリババクラウドの詳細は、こちらからご覧ください。
[アリババクラウドジャパン公式ページ](https://www.alibabacloud.com/ja)*