Python + AWS LambdaでSSL証明書の有効期限をチェックする

こんにちは！虎の穴ラボのNSSです。

みなさんの会社では、SSL証明書の有効期限をどのように管理しているでしょうか？

最近では、AWSやGCPなどのクラウドサービスが管理するSSL証明書を利用することで、
有効期限が切れる前に自動更新してくれるサービスもあります。
しかし、クラウドマネージドなSSL証明書が使用できない都合があったり、
利用しているクラウドサービスにSSL証明書の自動更新機能がなかったりすることがあります。

虎の穴ラボでは、今までスプレッドシートにドメイン名と有効期限を記載し、
Google Apps Script（GAS）を使って、有効期限がせまったドメインをSlack通知するという方法をとっていました。
一応これでもチェックすることはできますが、更新にあわせてスプレッドシートを手動で更新しなければならず、
忘れると通知が出続けてしまいます。
また、万が一スプレッドシート記載の日付が間違っていた場合、検知できず障害の原因となります。

SSL証明書の有効期限の確認は、その証明書自体にアクセスするのが一番確実です。
そこで、SSL証明書を直接チェックするツールを作成してみました。

1. 仕様・システム構成

今回、実装するチェックツールの主な仕様は次の通りです。

• 毎日AM10:00にSSL証明書の有効期限チェックをする
• SSL証明書の有効期限が近いドメインを検知してSlackにメッセージを通知する
• チェック対象のドメイン名、メッセージを通知し始める日数、SlackのWebHookURLは外部から設定できる

今回はCloudWatchをトリガーとしてAWS Lambda（以下Lambda）の関数を実行します。
システム構成は、次の図の通りです。

2. 実装

Python3.7を使って、Lambdaの関数を実装します。 実装するにあたり、どのようにPythonでSSL証明書の有効期限をチェックするか調べました。

結果、こちらの記事が非常に参考になりましたので、 こちらをもとに実装しました。 qiita.com

以下が実装した内容です。

import json
import datetime
import pytz
import os
import socket
import ssl
import requests
import sys

jst = pytz.timezone('Asia/Tokyo')

# 1. 実行ハンドラー：Lambdaはここから開始
def lambda_handler(event, context):
    domains = [x.strip() for x in str(os.getenv('Urls')).split(',')]
    webhook = os.getenv('IncommingWebhooks')
    try:
        # SSL期限チェック
        ssl_expires = {}
        for domain in domains:
            is_ssl_expires, ssl_expires_date = ssl_expires_in(domain)
            if is_ssl_expires:
                ssl_expires[domain] = ssl_expires_date
        print(len(ssl_expires))
        if len(ssl_expires) != 0:
            text = f"以下のSSL証明書が{os.getenv('BufferDays')}日以内に期限切れになります"
            for domain, expired_date in ssl_expires.items():
                text += f"\n{domain} - 有効期限 : {expired_date}"
            send_slack(text)
    except requests.RequestException as e:
        print(e)
        raise e
    except:
        print(sys.exc_info())
        text = "SSLチェック中にエラーが発生しました。"
        send_slack(text)
        raise


# 残日数の取得
def ssl_valid_time_remaining(hostname):
    expires = ssl_expiry_datetime(hostname)
    return expires - datetime.datetime.now(jst), expires

# SSLチェック関数
def ssl_expires_in(hostname):
    buffer_days=int(os.getenv('BufferDays'))
    try:
        remaining, expires = ssl_valid_time_remaining(hostname)
        return valid_date(remaining, expires, buffer_days)
    except:
        return True, "SSLチェックに失敗しました"

# 有効期限をジャッジする関数
def valid_date(remaining, expires, buffer_days):
    if remaining < datetime.timedelta(days=0):
        return True, "有効期限切れ、手遅れです"
    elif remaining < datetime.timedelta(days=buffer_days):
        return True, expires.strftime("%Y/%m/%d")
    else:
        return False, ""

# 有効期限の取り出し
def ssl_expiry_datetime(hostname):
    ssl_date_fmt = r'%b %d %H:%M:%S %Y %Z'
    utc_datetime = datetime.datetime.strptime(
        ssl_expiry(hostname), ssl_date_fmt)
    return utc_datetime.astimezone(jst)

# 有効期限の取得
def ssl_expiry(hostname):
    print("{}の接続を開始します。".format(hostname))
    context = ssl.create_default_context()
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) as sock:
        with context.wrap_socket(sock, server_hostname=hostname) as ssock:
            ssock.settimeout(3.0)
            ssock.connect((hostname, 443))
            ssl_info = ssock.getpeercert()
            ssock.close()
            print(ssl_info)
            return ssl_info['notAfter']  # ssl_info['notAfter'] が証明書の期限

# Slack通知
def send_slack(text):
    webhook = os.getenv('IncommingWebhooks')
    print('slack通知')
    requests.post(webhook, data=json.dumps({
        # 通知内容
        'text': text
    }))

実装の内容について解説します。

上記の実装のなかで、os.getenv()となっている部分が環境変数を参照している箇所です。
環境変数には、次のような値を指定します。

キー	値
Urls	SSLチェックをするドメイン名。カンマ区切りで複数指定。
BufferDays	有効期限を通知し始める日数
IncommingWebhooks	SlackのWebHookURL

環境変数で設定したドメインに対してSSLソケット通信を行い、JSON形式でSSL証明書の情報を取得します。
ソケット通信の詳しい解説はPython公式ドキュメントも合わせてご確認ください。

docs.python.org

取得したJSONのうち有効期限['notAfter']だけを取得します。

# 有効期限の取得
def ssl_expiry(hostname):
    print("{}の接続を開始します。".format(hostname))
    context = ssl.create_default_context()
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) as sock:
        with context.wrap_socket(sock, server_hostname=hostname) as ssock:
            ssock.settimeout(3.0)
            ssock.connect((hostname, 443))
            ssl_info = ssock.getpeercert()
            ssock.close()
            print(ssl_info)
            return ssl_info['notAfter']  # ssl_info['notAfter'] が証明書の期限

取得できた日付はJul 19 14:59:59 2020 GMTのような文字列になっているので、日付に変換して残り日数を計算します。
あとは算出した残り日数を判定してSlackに通知すれば完成です。
なお、今回はエラーとなった時「SSLチェックに失敗しました」とSlackに通知するようにしました。

CloudWatchの設定

関数の設定が終わったら、トリガーを設定します。
CloudWatchコンソールを開き、イベントの中のルールを選択します。
「ルールの作成」ボタンをクリックしてルール作成画面画面を開き、cron式を設定します。
毎日AM10:00通知する場合、日本時間はプラス9時間となるので、
0 1 * * ? *と設定します。
平日のみ通知する場合、
0 1 ? * 2-6 *と設定します。
またターゲットに先ほど作成した、Lambda関数を指定します。

時間通りにSlack通知が飛べば成功です。

まとめ

今回は、SSL証明書の有効期限チェックについてご紹介しました。
SSL証明書の更新は、１つ忘れただけでシステムが動かなくなってしまうこともあるので、
重要なことなのですが、検知しづらいのが問題です。
自動更新できれば言うことはないですが、いろいろな都合でできないと言う場合に今回の方法を試してみていただけると幸いです。

P.S.

初めて札幌で採用説明会を実施することになりました！近隣にお住みの方はぜひお申し込みください！

yumenosora.connpass.com

今年もLT会を開催します！

yumenosora.connpass.com

他にもカジュアル面談は随時受付中ですので、気になる方はぜひお申し込みください

yumenosora.connpass.com