AZURE AAD证书最佳实践监控

新钛云服 2024/07/04

需求:随着应用服务业务证书的增多,人为手动登录azure平台查看证书繁琐、用户权限控制、容易忘记登录等等信息的风险，这时就要自动化管理来提高安全性与可靠性。

Azure AD 证书管理工具可以帮助自动化证书的续期和监控。
通过这个工具,您可以设置证书的到期提醒,并自动完成证书的续期流程。
该工具可以显示证书的详细信息,并提供证书的使用情况报告。
设置证书到期警报,提醒您证书即将到期。
编写自动化脚本,定期扫描并报告 Azure AD 中所有证书的状态。

01 Microsoft Entra ID服务授权

登录azure网站输入地址 https://portal.azure.cn 打开服务Microsoft Entra ID服务，进入到应用注册板块

在API权限模块添加配置权限

API/权限名称:Application.Read.All 和授权类型:应用程序

证书和密码位置生成1个新客户端密码,这个是为了api调用时提取所有服务证书时间信息。

02 Postman调用测试接口返回

1、获取access_token信息

POST请求地址https://login.partner.microsoftonline.cn/目录(租户) ID/oauth2/v2.0/token

client_id和client_secret是新客户端id与密码信息

2、通过access_token获取证书截止期限

提取json日志输出返回信息可以使用谷歌浏览器插件JSON-handle查看更加直观

03 封装成CronJob跑任务输出LOG

1、编写Python脚本azure-aad_certdate.py文件

把步骤二Postman调用的信息封装成Python文件

import re
import requests
import json
import warnings
from datetime import datetime

warnings.filterwarnings("ignore")  # 忽略所有警告

#DATETIME_FORMAT = '%Y-%m-%d %H:%M:%S'  # 如果要日期也要时间，就把这一行取消注释
DATETIME_FORMAT = '%Y-%m-%d'  # 如果只要日期，就把这一行取消注释


def get_applications_info():
    # 获取 access_token
    token_url = "https://login.partner.microsoftonline.cn/xxxxxxxx-xxxxx-xxxx-xxxxx-xxxxxxxxxxxxx/oauth2/v2.0/token"
    token_payload = 'grant_type=client_credentials&client_id=xxxxxxxxx-4c7d-xxxxxx-8beb-xxxxxxxxx&client_secret=xxxxxxxxxxxh0wbS~VxxxxxxxxxG4y&scope=https%3A%2F%2Fmicrosoftgraph.chinacloudapi.cn%2F.default'
    token_headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Cookie': 'fpc=Akixxxxxxxxnf-7exxxxxxxxxxxx; stsservicecookie=estsfd; x-ms-gateway-slice=estsfd'
    }

    token_response = requests.request("POST", token_url, headers=token_headers, data=token_payload)
    token_data = json.loads(token_response.text)
    access_token = token_data['access_token']

    # 使用 access_token 发起 GET 请求
    url = "https://microsoftgraph.chinacloudapi.cn/v1.0/applications"
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Authorization': f'Bearer {access_token}'
    }

    response = requests.request("GET", url, headers=headers).json()
    return response.get("value", [])


def parse_time(time_str):
    # 正则表达式匹配三种时间格式
    pattern = re.compile(r'^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(\.\d{1,6})?(Z)?$')
    match = pattern.match(time_str)
    if not match:
        return ''
    _, ms, z = match.groups()
    format_string = '%Y-%m-%dT%H:%M:%S'
    if ms:
        format_string += '.%f'
    if z:
        format_string += 'Z'
    return datetime.strftime(datetime.strptime(time_str, format_string), DATETIME_FORMAT)


def format_data(values):
    nvalues = []
    for item in values:
        password_credentials = [
            dict(
                appId=item.get("appId"),
                displayName=item.get("displayName"),
                keyId=jtem.get("keyId"),
                type='passwordCredentials',
                startDateTime=parse_time(jtem.get("startDateTime")),
                endDateTime=parse_time(jtem.get("endDateTime"))
            )
            for jtem in item.get("passwordCredentials", [])]
        key_credentials = [
            dict(
                appId=item.get("appId"),
                displayName=item.get("displayName"),
                type='keyCredentials',
                keyId=jtem.get("keyId"),
                startDateTime=parse_time(jtem.get("startDateTime")),
                endDateTime=parse_time(jtem.get("endDateTime"))
            )
            for jtem in item.get("keyCredentials", [])]
        nvalues.extend(password_credentials)
        nvalues.extend(key_credentials)
    return nvalues


def main():
    values = get_applications_info()
    nvalues = format_data(values)
    for item in nvalues:
        print(json.dumps(item))


if __name__ == '__main__':
    main()

2、制作Dockerfile文件封装成镜像

#编译镜像

docker build -t xxxxxxx.xxxxxxxxx.cn/xxxxxx/azure-aad-certdate:v1 .

#推送镜像到镜像仓库

docker push xxxxxxx.xxxxxxxxx.cn/xxxxxx/azure-aad-certdate:v1

FROM python
ENV LANG=C.UTF-8
ENV TZ=Asia/Shanghai

RUN pip install pyyaml  --upgrade -i  https://pypi.tuna.tsinghua.edu.cn/simple
RUN pip install requests --upgrade -i https://pypi.tuna.tsinghua.edu.cn/simple
COPY azure-aad_certdate.py   /opt/

CMD ["sleep","999"]

3、封装的镜像编写成Cronjob yaml文件

命令执行kubectl apply -f azure_aad_certdate.yaml

apiVersion: batch/v1beta1
kind: CronJob
metadata:
  name: azure-aad-certdate-monitor-server
  namespace: monitoring
spec:
  schedule: "* 9 * * *"
  concurrencyPolicy: Forbid
  jobTemplate:
    spec:
      parallelism: 1      #作业的最大并行度，默认为1
      completions: 1      #期望的成功完成的作业次数，成功运行结束的Pod数量
      ttlSecondsAfterFinished: 600   #终止状态作业的生存时长，超期将被删除
      backoffLimit: 3                #将作业标记为Failed之前的重试次数
      activeDeadlineSeconds: 60      #作业启动后可处于活动状态的时长
      template:
        spec:
          containers:
          - name: azure-aad-certdate-monitor
            image: xxxxxxx.xxxxxxxxx.cn/xxxxxx/azure-aad-certdate:v1
            #imagePullSecrets:
            imagePullPolicy: Always
            command:
            - /bin/sh
            - -c
            - python /opt/azure-aad_certdate.py
          restartPolicy: OnFailure
  startingDeadlineSeconds: 300   #因错过时间点而未执行的作业的可超期时长

4、查看Cronjob输出pod日志

azure-aad-certdate-monitor-server pod的json日志输出详情

04 SLS配置监控

1、编写sls查询语法进行查询

#查询证书时间在0至30天之内到期的服务信息

((_namespace_ : monitoring and _container_name_: azure-aad-certdate-monitor))| select DISTINCT appId,displayName,type,startDateTime ,endDateTime,date_diff('day', date_parse(split(_time_, 'T')[1], '%Y-%m-%d'), date_parse(endDateTime, '%Y-%m-%d')) as days having days < 30 and days > 0 ORDER BY days ASC LIMIT 1000