使用 Python 从基于 UserStatus 和 LastSuccessfulLoginDate 的 OCI IAM 身份域中删除非活动用户

简介

Oracle Cloud Infrastructure (OCI) 是一个云平台，能够提供存储、网络和基础设施等一系列云服务。Oracle Cloud Infrastructure Identity and Access Management (OCI IAM) 是一项支持您管理 OCI 资源访问的服务。它为用户和组提供验证和授权。

基于在 OCI IAM 身份域中上次成功登录日期删除非活动用户在 OCI 每用户定价模型的上下文中尤其有用，原因如下：

成本优化： OCI 采用按用户定价模式，对每个用户账户计费。通过定期识别和删除非活动用户，您可以显著降低订阅成本。随着组织规模和用户数量的增长，成本优化变得更加重要。
安全增强：非活动用户账户存在潜在的安全风险。这些帐户可能不再被合法用户使用，但如果无人看管，它们仍可能被恶意行为者利用。删除非活动用户可确保只有授权人员才能访问您的 OCI 资源，从而减少攻击面并增强整体安全性。
Resource Management：管理大量非活动用户帐户可能会带来管理上的负担。它会打乱您的 IAM 环境，使您更难专注于活动用户账户。自动识别和删除不活动用户的流程，释放管理资源，让您的团队专注于更具战略性的任务。
合规性：合规性标准通常要求组织定期审查和管理用户账户。删除非活动用户可以帮助您满足合规性要求并展示对数据安全和监管的承诺。

为了有效地实现这个过程，下面的 python 代码片段可用于根据非活动用户的上次成功登录日期和活动/非活动状态自动识别和删除非活动用户。通过定期运行此类脚本，您可以在 OCI 中保持精简、安全的 IAM 环境，同时优化成本。

目标读者

IAM 专业人员和管理员。

目标

使用带 Python 的 REST API，基于用户状态和 lastSuccessfulLoginDate 从 OCI IAM 身份域中删除用户。

先决条件

活动的 OCI 订阅。
熟悉 OCI IAM 和 Python。
需要具备使用 OCI IAM 身份域 REST API 的知识。
具有管理应用程序的授权的 IAM 用户（身份域管理员、安全管理员或应用程序管理员）。
系统上安装了 Python 3.x。
已安装 urllib3、requests 和 datetime Python 软件包。

任务 1：在 OCI IAM 身份域中创建机密应用程序

按照 Oracle Identity Cloud Service：第一个 REST API 调用创建机密应用程序并检索客户端 ID 和客户端密钥，然后可以使用这些客户端密钥对 OCI IAM 执行 REST API 调用，以检索访问令牌和后续 API 端点。

任务 2：设置 `config.json` 文件

在本地计算机上设置配置文件。config.json 文件包含有关用于生成访问令牌的身份域 URL、客户端 ID 和客户端密钥的信息。

{
"iamurl"         : "https://idcs-###########.identity.oraclecloud.com",
"client_id"      : "#######################",
"client_secret"  : "#######################"
}

任务 3：获取访问令牌

config.json 文件就位后，您需要做的第一件事是生成访问令牌，该令牌可用于对 OCI IAM 端点进行进一步的 REST API 调用。

在下面的代码片段中，函数 get_encoded 将客户机 ID 和客户机密钥作为参数并返回 base64-encoded 字符串。此编码字符串作为参数进一步传递到函数 get_access_token 作为授权标头，以便通过执行 POST 请求来获取访问令牌。

#get base64 encoded
def get_encoded(self,clid, clsecret):
    encoded = clid + ":" + clsecret
    baseencoded = base64.urlsafe_b64encode(encoded.encode('UTF-8')).decode('ascii')
    return baseencoded

#get access token
def get_access_token(self,url, header):
    para = "grant_type=client_credentials&scope=urn:opc:idm:__myscopes__"
    response = requests.post(url, headers=header, data=para, verify=False)
    jsonresp = json.loads(response.content)
    access_token = jsonresp.get('access_token')
    return access_token

#print access token
def printaccesstoken(self):
    obj = IAM()
    encodedtoken = obj.get_encoded(clientID, clientSecret)
    extra = "/oauth2/v1/token"
    headers = {'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8',
               'Authorization': 'Basic %s' % encodedtoken, 'Accept': '*/*'}
    accesstoken = obj.get_access_token(idcsURL + extra, headers)
    return accesstoken

任务 4：处理一批用户信息（lastSuccessfulLoginDate、userStatus）以清除用户

我们具有访问令牌，通过该令牌，我们可以对不同的 OCI IAM 身份域 REST 端点进行进一步的 REST API 调用。下面的代码段显示了如何向 /admin/v1/Users 端点发出带有必要标头和参数的 GET 请求，以便检索用户总数，然后根据结果总数和每个请求的计数计算所需的迭代次数。

该函数获取 YYYY-MM-DD 格式的当前日期，并将其存储在变量 current_date 中，然后迭代每个用户数据。
在每次迭代中，它会发送带适当参数的 GET 请求以检索一批用户信息 (username，lastSuccessfulLoginDate and active) ，然后执行一些条件检查。
然后，代码基于定义的条件检查，分别为 username.txt 和 userid.txt 文件中的所有匹配用户附加用户名和用户 ID，并将其标记为删除。首先，它检查用户状态，如果为 False （即 inactive ），则会将用户附加到 txt 文件并继续执行循环的下一次迭代。然后，它会检查用户状态，如果是无，则会输出 _{user} 从未访问过的内容。决定在控制台上手动删除，然后继续执行循环的下一次迭代。
该脚本检索用户的上次成功登录日期，计算自上次登录以来的天数，并将其打印到控制台。对于任何未登录超过 90 天的用户，脚本会继续检查用户是否具有与其关联的特定管理员角色。任何 90 天没有登录并且具有无管理员角色的用户都会被标识并标记为删除。
处理所有用户后，一组要删除的用户 ID 将写入 JSON 文件 username.json。批量删除请求是通过格式化特定 JSON 格式的数据来准备的。然后将批量删除请求发送到 /admin/v1/Bulk 端点，并输出一条消息以指示已标识的用户已被删除。

经过所有的迭代，函数返回到 main 函数。

注：有一个代码部分添加了有关未来操作或手动决策的说明，例如处理从未访问系统的用户。有时，管理员可能无法访问 OCI IAM，您可能不想删除它们。

def get_successfullogindate(self):
    extra = "/admin/v1/Users"
    obj = IAM()
    accesstoken = obj.printaccesstoken()
    headers = {'Authorization': 'Bearer ' + accesstoken}
    resp = requests.get(idcsURL+extra, headers=headers, verify=False)
    jsonresp = json.loads(resp.content)
    totalCount = jsonresp.get("totalResults")
    print("Total number of users: " + str(totalCount))

    current_date = datetime.now().date()
    startIndex = 1
    count = 50
    loop = int(totalCount / count)
    extra2 = "/admin/v1/AppRoles"

    for i in range(loop + 1):
        param = {
            'attributes': "username,active,urn:ietf:params:scim:schemas:oracle:idcs:extension:userState:User:lastSuccessfulLoginDate",
            'startIndex': startIndex, 'count': count}
        resp = requests.get(idcsURL+extra, headers=headers, verify=False, params=param)
        startIndex += count
        jsonresp = json.loads(resp.content)

        tempjsn = jsonresp.get("Resources")

        for x in tempjsn:
            trimjsn = {}
            username = trimjsn['userName'] = x.get("userName")
            userid = x.get("id")
            userStatus = x.get("active")
            userState = x.get("urn:ietf:params:scim:schemas:oracle:idcs:extension:userState:User")
            if userStatus is False:
                print(username + " is in inactive state. Will be deleted.")
                with open('username.txt', 'a') as c:
                    c.write(username)
                    c.write('\n')
                with open('userID.txt', 'a') as d:
                    d.write(userid)
                    d.write('\n')
                continue
            if userState is None:
                print(username + " has never accessed. Kindly make a decision to delete it manually on console")
                # with open('username.txt', 'a') as c:
                #     c.write(username)
                #     c.write('\n')
                # with open('userID.txt', 'a') as d:
                #     d.write(userid)
                #     d.write('\n')

                #  ** Uncomment the lines commented above to delete the users who have never accessed **

                continue
            lastLoginDate = x['urn:ietf:params:scim:schemas:oracle:idcs:extension:userState:User']['lastSuccessfulLoginDate']
            target_dates = datetime.strptime(lastLoginDate, '%Y-%m-%dT%H:%M:%S.%fZ').date()
            num_days = (current_date - target_dates).days
            #print(username + " Last logged in " + str(num_days) + " days ago")
            if num_days > 30:
                #print("entered if condition for more than 90 days ")
                params2 = {"filter": f'members[value eq "{userid}"] and app.value eq "IDCSAppId" and (displayName eq "Identity Domain Administrator" or displayName eq "Security Administrator" or displayName eq "Application Administrator" or displayName eq "User Administrator" or displayName eq "User Manager" or displayName eq "Help Desk Administrator" or displayName eq "Audit Administrator")'}
                resp2 = requests.get(idcsURL+extra2, headers=headers, verify=False, params=params2)
                jsonresp3 = json.loads(resp2.content)
                totalCount1 = jsonresp3.get("totalResults")
                if totalCount1 < 1:
                    print(username + " has not Logged in for 90 days and has NO admin role. Will be deleted.")
                    with open('username.txt', 'a') as c:
                        c.write(username)
                        c.write('\n')
                    with open('userID.txt', 'a') as d:
                        d.write(userid)
                        d.write('\n')

    try:
        with open('userID.txt','r') as g:
            lines = len(g.readlines())
    except FileNotFoundError:
        print("No users found to be Deleted by the Script")
        exit()

    x=0
    while x<lines:
        with open('userID.txt','r') as e:
            content = e.readlines()
            id=content[x]
            id=id.rstrip(id[-1])
        paradelete=json.dumps(
                {
                "method": "DELETE",
                "path": "/Users/"+id+"?forceDelete=true"
                }
                )
        with open('username.json', 'a') as j:
                j.writelines(paradelete)
                j.write(',')

        x=x+1
    with open('username.json', 'r') as file:
        data = file.read()
    with open('formatedData.json', 'a') as file1:
        file1.writelines('{ \n')
        file1.writelines('"schemas": ["urn:ietf:params:scim:api:messages:2.0:BulkRequest"],\n')
        file1.writelines('"Operations": [\n')
        data2=data.rstrip(data[-1])
        file1.write(data2)
        file1.write(']}')

    headers = {'Authorization': 'Bearer ' + accesstoken}
    bulkdata = json.load(open('formatedData.json'))
    param = {'forceDelete': True}
    payload = json.dumps(bulkdata)
    headers2 = {'Content-Type': 'application/json','Authorization': 'Bearer ' + accesstoken, 'Accept': '*/*'}
    extra2="/admin/v1/Bulk"
    #respdelete = requests.request("POST",idcsURL + extra2, headers=headers2, verify=False, params=param,data=payload)
    print("The identified users has been deleted ")

    try:
        os.remove('userID.txt')
        os.remove('username.txt')
        os.remove('formatedData.json')
        os.remove('username.json')
    except FileNotFoundError:
        exit()

任务 5：在 OCI Cloud Shell 中使用脚本

脚本准备就绪后，可以在本地计算机上（安装了 Python）或任何支持 Python 开发的 IDE 上轻松执行该脚本。我们使用 OCI Cloud Shell 运行脚本并获取所需报告。

登录到 OCI 控制台，从屏幕右上角打开 Cloud Shell，然后上载 Python 脚本和 config.json 文件。
执行 python DeleteUser_PriorCheck_InactivePeriod60days_IsAdmin.py。

注：DeleteUser_PriorCheck_InactivePeriod60days_IsAdmin.py 是使用以上代码片段开发的 Python 脚本。

确认

作者 - Gautam Mishra（高级云工程师）
贡献者 - Chetan Soni（云解决方案工程师）