#!/usr/bin/env python3 """Zero Trust 정리 — 기기 등록 · 좌석 · 2차 인증(MFA) (운영 매뉴얼 M13). **기본이 미리보기입니다.** 실제로 바꾸려면 `--apply` 를, 다섯 건을 넘기면 `--yes` 도 붙입니다. 표준 라이브러리만 씁니다 — 설치할 것이 없습니다. export CF_API_TOKEN=<토큰> # 권한: Access Read+Write · Devices Read+Write export CF_ACCOUNT_ID=<계정 ID> # Zero Trust 화면 주소의 32자리 python cleanup.py report # 읽기만 — 좌석·기기·중복 요약 python cleanup.py devices --stale-days 90 # 미리보기 python cleanup.py devices --stale-days 90 --apply --yes python cleanup.py devices --keep-newest # 사람별 최신 1대만 남깁니다 python cleanup.py devices --stale-days 180 --csv 대상.csv # 검토용 목록만 뽑습니다 python cleanup.py seats --inactive-days 60 python cleanup.py seats --inactive-days 180 --csv 좌석.csv python cleanup.py mfa --email 본인@주소 [--apply] 토큰은 인자로 받지 않습니다 — 셸 히스토리에 남기지 않기 위해서입니다. 알고 시작해야 하는 함정 1. active_registrations=include 를 빼면 「등록 0개」 줄이 통째로 안 보입니다(M3). 2. page= 는 2750 not supported 로 죽습니다. **커서 페이징**입니다. 3. 등록을 지우면 그 단말이 재인증할 때 **줄이 새로 생깁니다**(M4). 4. 좌석 반납은 차단이 아닙니다. 다시 로그인하면 다시 점유합니다(M6). 5. **MFA 목록 API 가 없습니다.** last_seen_identity 안의 mfa_authenticators 가 유일한 창구이고, 그것은 「마지막 로그인 순간에 찍힌 사진」입니다 — 그 뒤에 등록한 수단은 안 보입니다. 6. 삭제와 취소(revoke)는 다릅니다. 취소는 가상 IP 를 회수하지 않습니다(M5). 공식 문서 등록 목록 https://developers.cloudflare.com/api/resources/zero_trust/subresources/devices/subresources/registrations/methods/list/ 등록 삭제 https://developers.cloudflare.com/api/resources/zero_trust/subresources/devices/subresources/registrations/methods/delete/ 기기 목록 https://developers.cloudflare.com/api/resources/zero_trust/subresources/devices/subresources/devices/methods/list/ 사용자 https://developers.cloudflare.com/api/resources/zero_trust/subresources/access/subresources/users/methods/list/ 좌석 반납 https://developers.cloudflare.com/api/resources/zero_trust/subresources/seats/methods/edit/ 마지막 신원 https://developers.cloudflare.com/api/resources/zero_trust/subresources/access/subresources/users/subresources/last_seen_identity/methods/get/ 토큰 만들기 https://developers.cloudflare.com/fundamentals/api/get-started/create-token/ """ import argparse, csv, json, os, sys, urllib.parse, urllib.request from datetime import datetime, timezone API = "https://api.cloudflare.com/client/v4" TOKEN = os.environ.get("CF_API_TOKEN", "") ACCOUNT = os.environ.get("CF_ACCOUNT_ID", "") if not TOKEN or not ACCOUNT: sys.exit("CF_API_TOKEN 과 CF_ACCOUNT_ID 를 환경변수로 넣어 주세요.") def api(method, path, body=None, params=None): url = API + "/accounts/" + ACCOUNT + path + ("?" + urllib.parse.urlencode(params) if params else "") data = json.dumps(body).encode("utf-8") if body is not None else None req = urllib.request.Request(url, data=data, method=method) req.add_header("Authorization", "Bearer " + TOKEN) if data: req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, timeout=30) as r: j = json.loads(r.read().decode("utf-8")) except urllib.error.HTTPError as e: sys.exit("HTTP %s — %s" % (e.code, e.read().decode("utf-8", "replace")[:200])) if not j.get("success", True): sys.exit("; ".join(x.get("message", "") for x in (j.get("errors") or []))) return j.get("result") def get_all(path, params=None): """커서 페이징. page= 를 쓰면 2750 으로 죽으므로 절대 쓰지 않습니다.""" out, cursor = [], "" while True: q = dict(params or {}) q["per_page"] = "100" if cursor: q["cursor"] = cursor url = API + "/accounts/" + ACCOUNT + path + "?" + urllib.parse.urlencode(q) req = urllib.request.Request(url) req.add_header("Authorization", "Bearer " + TOKEN) with urllib.request.urlopen(req, timeout=30) as r: j = json.loads(r.read().decode("utf-8")) out.extend(j.get("result") or []) nxt = (j.get("result_info") or {}).get("cursor") or "" if not nxt or nxt == cursor or not j.get("result"): return out cursor = nxt def email_of(reg): for k in ("user", "last_seen_user"): v = (reg.get(k) or {}).get("email") if v: return v.lower() return "" def age_days(v): if not v: return None try: d = datetime.fromisoformat(str(v).replace("Z", "+00:00")) except ValueError: return None if not d.tzinfo: d = d.replace(tzinfo=timezone.utc) return (datetime.now(timezone.utc) - d).days def guard(items, args, what): """실행 조건. 미리보기가 기본이고, 여러 건을 한꺼번에 바꿀 때는 --yes 를 더 받습니다.""" if not items: print("\n대상 없음 — 바꿀 것이 없습니다.") return False if not args.apply: print("\n미리보기입니다. 실제로 %s 하려면 --apply 를 붙이십시오." % what) return False if len(items) > 5 and not args.yes: print("\n대상이 %d건입니다. 다섯 건을 넘으면 --yes 를 함께 붙여야 합니다." % len(items)) return False return True def cmd_report(args): users = api("GET", "/access/users", params={"per_page": "200"}) or [] regs = get_all("/devices/registrations", {"status": "all"}) devs = get_all("/devices/physical-devices", {"active_registrations": "include"}) seated = [u for u in users if u.get("access_seat")] print("사용자 %d명 · 좌석 점유 %d명" % (len(users), len(seated))) print("기기 줄 %d개 · 등록 %d건" % (len(devs), len(regs))) zero = 0 for d in devs: ar = d.get("active_registrations") if (len(ar) if isinstance(ar, list) else (ar or 0)) == 0: zero += 1 print("등록 0개인 기기 줄 %d개 (include 없이 조회하면 안 보이는 줄입니다)" % zero) by = {} for r in regs: by.setdefault(email_of(r) or "(사람 미상)", []).append(r) many = sorted(((k, v) for k, v in by.items() if len(v) > 1), key=lambda kv: -len(kv[1])) if many: print("\n등록이 여러 건인 사람 %d명 — 「지울수록 목록이 길어진다」(M4)" % len(many)) for email, rows in many[:10]: print(" %s — %d건" % (email, len(rows))) print("\n좌석을 물고 있는데 오래 안 들어온 사람:") idle = [(age_days(u.get("last_successful_login")), u) for u in seated] idle = [(a, u) for a, u in idle if a is not None and a >= 30] for a, u in sorted(idle, reverse=True)[:10]: print(" %s — %d일 전 로그인" % (u.get("email"), a)) if not idle: print(" 없음(30일 기준)") def write_csv(path, header, rows): """검토용 목록을 파일로 냅니다 — **미리보기에서도 씁니다.** 왜 있나 — 정리는 대개 「우리가 지운다」가 아니라 「목록을 뽑아 부서에 돌려 확인받는다」로 시작합니다. 화면 출력은 그 용도로 못 씁니다. 🔴 인코딩은 **utf-8-sig** 입니다. 그냥 utf-8 로 쓰면 Excel 이 한글을 깨뜨립니다 (엑셀은 BOM 이 없으면 시스템 인코딩으로 읽습니다). newline="" 은 줄이 한 칸씩 벌어지는 것을 막습니다 — 윈도우에서만 나는 증상입니다. """ with open(path, "w", encoding="utf-8-sig", newline="") as fh: w = csv.writer(fh) w.writerow(header) w.writerows(rows) print("\n검토용 목록을 넣었습니다: %s (%d줄)" % (path, len(rows))) def cmd_devices(args): regs = get_all("/devices/registrations", {"status": "all"}) if args.email: regs = [r for r in regs if email_of(r) == args.email.lower()] rows = [((age_days(r.get("last_seen") or r.get("created")) or -1), r) for r in regs] if args.stale_days is not None: targets = [r for a, r in rows if a >= args.stale_days] what = "%d일 이상 안 쓴 등록" % args.stale_days elif args.keep_newest: # 🔴 사람을 못 읽은 줄은 대상에서 뺍니다. 빈 문자열로 묶으면 서로 다른 사람의 # 등록이 한 덩어리가 되어 **남의 것을 지웁니다.** by, skipped = {}, 0 for a, r in rows: who = email_of(r) if not who: skipped += 1 continue by.setdefault(who, []).append((a, r)) targets = [] for lst in by.values(): lst.sort(key=lambda x: x[0]) # 나이 오름차순 = 최신이 앞 targets.extend(r for _, r in lst[1:]) what = "사람별 최신 1건을 뺀 나머지" if skipped: print("사람을 못 읽은 등록 %d건은 대상에서 뺐습니다(누구 것인지 모르면 지우지 않습니다)." % skipped) else: sys.exit("무엇을 고를지 지정해야 합니다: --stale-days N 또는 --keep-newest") print("등록 %d건 중 대상 %d건 — %s" % (len(regs), len(targets), what)) for r in targets: print(" %s %s" % (email_of(r) or "(사람 미상)", r.get("name") or "-")) print("\n⚠️ 지운 단말이 재인증하면 줄이 새로 생깁니다(M4). 목록을 줄이는 것이 목적이면") print(" 등록 정책·IdP 계정까지 함께 봐야 합니다 — 이 코드는 등록만 지웁니다.") if args.csv: idle = {id(r): a for a, r in rows} write_csv(args.csv, ["이메일", "단말 이름", "등록 id", "마지막 접속", "안 쓴 일수"], [[email_of(r) or "", r.get("name") or "", r.get("id") or "", r.get("last_seen") or r.get("created") or "", idle.get(id(r), -1)] for r in targets]) if not guard(targets, args, "삭제"): return for r in targets: api("DELETE", "/devices/registrations/" + str(r.get("id"))) print(" 지웠습니다: %s" % r.get("id")) print("\n삭제 %d건" % len(targets)) def cmd_seats(args): users = api("GET", "/access/users", params={"per_page": "200"}) or [] seated = [u for u in users if u.get("access_seat") and u.get("seat_uid")] if args.email: seated = [u for u in seated if str(u.get("email", "")).lower() == args.email.lower()] if args.inactive_days is not None: seated = [u for u in seated if (age_days(u.get("last_successful_login")) or -1) >= args.inactive_days] if not args.email and args.inactive_days is None: sys.exit("무엇을 고를지 지정해야 합니다: --email 또는 --inactive-days N") print("좌석 반납 대상 %d명" % len(seated)) for u in seated: print(" %s seat_uid=%s" % (u.get("email"), u.get("seat_uid"))) print("\n⚠️ 반납은 차단이 아닙니다 — 다시 로그인하면 다시 점유합니다(M6).") print(" 이 요청은 대시보드 「Remove users」 버튼이 보내는 것과 같습니다.") if args.csv: write_csv(args.csv, ["이메일", "seat_uid", "마지막 로그인", "안 쓴 일수"], [[u.get("email") or "", u.get("seat_uid") or "", u.get("last_successful_login") or "", age_days(u.get("last_successful_login")) if u.get("last_successful_login") else -1] for u in seated]) if not guard(seated, args, "반납"): return api("PATCH", "/access/seats", body=[{"seat_uid": u.get("seat_uid"), "access_seat": False, "gateway_seat": False} for u in seated]) print("\n반납 %d건 보냈습니다." % len(seated)) def cmd_mfa(args): users = api("GET", "/access/users", params={"per_page": "200"}) or [] target = next((u for u in users if str(u.get("email", "")).lower() == args.email.lower()), None) if not target: sys.exit("그 주소의 사용자가 없습니다: " + args.email) uid = str(target.get("id") or "") r = api("GET", "/access/users/" + urllib.parse.quote(uid) + "/last_seen_identity") or {} lst = r.get("mfa_authenticators") or [] at = r.get("iat") print("사용자 %s" % target.get("email")) print("MFA 수단 %d개" % len(lst)) print("⚠️ 이 목록은 「마지막 로그인 순간에 찍힌 사진」입니다" + (" 찍힌 시각 " + datetime.fromtimestamp(at, timezone.utc).isoformat() if at else "")) print(" 그 뒤에 등록한 수단은 안 보입니다. 전용 목록 API 가 없어 이 방법뿐입니다(M9).") for m in lst: print(" %s type=%s id=%s" % (m.get("name") or m.get("type"), m.get("type"), m.get("id"))) print("\n⚠️ 지우면 그 사람은 다음 접속에 「Setup MFA」를 받습니다. 복구 코드는 없습니다(M9).") if not guard(lst, args, "삭제"): return for m in lst: api("DELETE", "/access/users/" + uid + "/mfa_authenticators/" + str(m.get("id"))) print(" 지웠습니다: %s" % (m.get("name") or m.get("type"))) p = argparse.ArgumentParser(description="Zero Trust 정리. 기본은 미리보기이고 --apply 로만 바뀝니다.") sub = p.add_subparsers(dest="cmd", required=True) common = argparse.ArgumentParser(add_help=False) common.add_argument("--apply", action="store_true", help="실제로 바꿉니다(없으면 미리보기)") common.add_argument("--yes", action="store_true", help="다섯 건을 넘길 때 함께 붙입니다") # 미리보기에서도 씁니다 — 검토를 받으려면 파일이 있어야 합니다. common.add_argument("--csv", metavar="파일", help="대상 목록을 CSV 로 냅니다(Excel 에서 바로 열립니다)") sub.add_parser("report", help="읽기만 — 좌석·기기·중복 요약").set_defaults(fn=cmd_report) d = sub.add_parser("devices", parents=[common], help="기기 등록 정리") d.add_argument("--stale-days", type=int) d.add_argument("--keep-newest", action="store_true") d.add_argument("--email") d.set_defaults(fn=cmd_devices) s = sub.add_parser("seats", parents=[common], help="좌석 반납") s.add_argument("--inactive-days", type=int) s.add_argument("--email") s.set_defaults(fn=cmd_seats) m = sub.add_parser("mfa", parents=[common], help="MFA 수단 조회·삭제") m.add_argument("--email", required=True) m.set_defaults(fn=cmd_mfa) a = p.parse_args() a.fn(a)