被裁员工删掉 96 个政府数据库——离岗访问权为什么必须秒级收回

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:ListBucket"],
      "Resource": ["arn:aws:s3:::gov-data-bucket", "arn:aws:s3:::gov-data-bucket/*"],
      "Condition": {
        "DateLessThan": {
          "aws:CurrentTime": "2025-08-01T00:00:00Z"
        },
        "DateGreaterThan": {
          "aws:CurrentTime": "2025-01-15T00:00:00Z"
        }
      }
    }
  ]
}

"""
offboard_revoke.py — 离岗时自动回收数字凭证
运行前设置环境变量：
  export SLACK_ADMIN_TOKEN=xoxp-xxx
  export GITHUB_ADMIN_TOKEN=ghp_xxx
  export AWS_PROFILE=gov-admin
  export DB_ADMIN_CONN="postgresql://admin:pass@db-host:5432/govdb"
"""

import os
import sys
import boto3
import requests
from datetime import datetime

def revoke_aws_access(username: str):
    """删除 IAM 用户，挂起所有关联 access key"""
    iam = boto3.client("iam")
    # 列出并删除所有 access key
    keys = iam.list_access_keys(UserName=username)
    for key in keys["AccessKeyMetadata"]:
        iam.delete_access_key(UserName=username, AccessKeyId=key["AccessKeyId"])
        print(f"[AWS] 已删除 access key: {key['AccessKeyId']}")
    # 删除用户
    iam.delete_user(UserName=username)
    print(f"[AWS] 已删除 IAM 用户: {username}")

def revoke_slack_access(email: str):
    """禁用 Slack 账号"""
    token = os.environ["SLACK_ADMIN_TOKEN"]
    # 先查找用户 ID
    resp = requests.get(
        "https://slack.com/api/users.lookupByEmail",
        headers={"Authorization": f"Bearer {token}"},
        params={"email": email},
    )
    user_id = resp.json()["user"]["id"]
    # 禁用账号
    requests.post(
        "https://slack.com/api/admin.users.disable",
        headers={"Authorization": f"Bearer {token}"},
        json={"user_id": user_id},
    )
    print(f"[Slack] 已禁用账号: {email}")

def revoke_db_access(username: str):
    """撤销数据库角色权限并删除用户"""
    import psycopg
    conn_str = os.environ["DB_ADMIN_CONN"]
    with psycopg.connect(conn_str) as conn:
        with conn.cursor() as cur:
            cur.execute(f"REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA public FROM {username};")
            cur.execute(f"REVOKE ALL PRIVILEGES ON ALL SEQUENCES IN SCHEMA public FROM {username};")
            cur.execute(f"DROP ROLE IF EXISTS {username};")
            print(f"[DB] 已撤销权限并删除角色: {username}")

def offboard(employee_id: str, username: str, email: str):
    """离岗流水线：依次回收各系统凭证"""
    print(f"=== 离岗凭证回收开始 {datetime.now()} ===")
    print(f"员工: {employee_id} / {username} / {email}")
    try:
        revoke_aws_access(username)
        revoke_slack_access(email)
        revoke_db_access(username)
    except Exception as e:
        print(f"[ERROR] 凭证回收失败: {e}")
        sys.exit(1)
    print(f"=== 离岗凭证回收完成 {datetime.now()} ===")

if __name__ == "__main__":
    # 实际部署中，这些参数应来自 HR 事件消息（如 Kafka topic / webhook）
    offboard(
        employee_id=sys.argv[1],
        username=sys.argv[2],
        email=sys.argv[3],
    )

python offboard_revoke.py EMP-00421 akhter akhter@gov.example.com

# 只允许数据运维组执行读取和备份，禁止删除
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  namespace: gov-data
  name: data-ops-reader
rules:
  - apiGroups: [""]
    resources: ["pods", "pods/log"]
    verbs: ["get", "list"]
  - apiGroups: ["batch"]
    resources: ["jobs"]
    verbs: ["get", "list", "create"]  # 允许创建备份 job，但不允许删除
---
# 删除操作需要单独的高权限角色，且绑定到极少数人
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  namespace: gov-data
  name: data-ops-admin
rules:
  - apiGroups: [""]
    resources: ["pods"]
    verbs: ["delete"]  # 仅此角色可删 pod
  - apiGroups: ["batch"]
    resources: ["jobs"]
    verbs: ["delete"]

--audit-log-path=/var/log/kubernetes/audit.log
--audit-log-maxage=30
--audit-policy-file=/etc/kubernetes/audit-policy.yaml

apiVersion: audit.k8s.io/v1
kind: Policy
rules:
  - level: RequestResponse
    verbs: ["delete", "deletecollection"]
    resources:
      - group: ""
        resources: ["pods", "secrets", "configmaps"]
      - group: "batch"
        resources: ["jobs", "cronjobs"]