服务账号+域级授权：跨域用户管理API批量操作实战

1. 这套方案到底在解决什么问题

但凡手里管着不止一个业务域名的IT同行，大概率都经历过这种场面：公司主体下面挂着三四个甚至十几个域名，每个域名都有自己的用户体系，新员工入职要在A域名开号，调岗又要去B域名建账号，离职了还得挨个域名去禁用。更头疼的是，季度审计要拉一份全公司的人员清单，你得登录N个管理后台，导出N份CSV，再手动合并去重。这种重复劳动做一次两次还能忍，做成常态就是纯纯的消耗。

这套方案的核心思路，就是用一个服务账号（Service Account）打通所有域名的管理权限，再通过API批量操作，把跨域用户管理这件事从"人肉登录后台"变成"跑一段脚本"。服务账号在这里扮演的角色，类似于一把万能钥匙——它不属于任何一个具体的人，而是属于一个应用身份，通过域级授权（Domain-Wide Delegation）获得模拟任意用户的能力，从而在多个域名之间自由穿梭。

具体能做什么？我列几个最典型的场景：跨域批量创建用户、统一重置密码、批量查询所有域名的用户列表、自动同步组织架构、定期清理僵尸账号。适合谁来参考？如果你是企业IT管理员、系统集成工程师，或者正在做身份管理（IAM）相关项目的开发者，这套东西能帮你省下大量重复操作的时间。哪怕你只管理一个域名，理解服务账号加API这套组合拳，对后续做自动化运维也是实打实的基础。

我先把话说在前面：这套方案的门槛不在代码本身，而在权限配置和API启用这两个环节。很多人卡在"明明代码没问题，就是报403"的阶段，根源基本都是授权没配对。下面我会把每个环节拆开讲透，包括我踩过的坑。

2. 整体架构与方案选型思路

2.1 为什么选服务账号而不是OAuth用户授权

管理多个域名的用户，最直觉的做法是用自己的管理员账号走OAuth授权，拿一个refresh token去调API。但这个方案有个致命问题：token是绑定在具体某个人身上的。一旦这个人离职、改密码、或者账号被临时锁定，所有自动化脚本全部瘫痪。而且OAuth的用户授权流程需要人工在浏览器里点同意，没法做到完全无人值守。

服务账号就不一样了。它是应用级别的身份，不依赖任何自然人。你创建好服务账号、下载好密钥文件（JSON格式），配置好域级授权，之后脚本就能7×24小时稳定运行，不需要任何人介入。密钥文件可以安全地存放在配置管理系统里，轮换也方便。

注意：服务账号的密钥文件是敏感凭证，绝对不能提交到代码仓库。我见过有人把JSON密钥直接push到公开仓库，结果被扫描工具抓到，整个域的用户数据面临泄露风险。正确做法是放在环境变量或专用的密钥管理服务里。

2.2 域级授权的工作原理

域级授权（Domain-Wide Delegation，简称DWD）是这套方案的关键机制。默认情况下，服务账号只能访问自己创建的资源，碰不到任何用户数据。开启DWD之后，超级管理员在管理控制台里给服务账号授予特定的API权限范围（Scope），服务账号就能"模拟"域内任意用户来执行操作。

打个比方：服务账号本身是一个没有身份的外部人员，DWD相当于超级管理员给它发了一张通行证，通行证上写明了"你可以代表本域内任何用户去访问这些API"。通行证上的权限范围是可以精确控制的，比如只给读写用户的权限，不给访问邮件的权限，这样即使密钥泄露，损失也能控制在有限范围内。

2.3 多域名管理的架构设计

管理多个域名时，架构上有两种选择：一是每个域名单独建一个服务账号，二是用一个服务账号管理所有域名。我推荐后者，原因很简单——维护成本低。一个服务账号只需要一套密钥、一份授权配置，脚本里通过切换subject参数就能操作不同域名。

具体实现上，服务账号的subject字段填哪个域的管理员邮箱，操作就作用在哪个域。比如subject填admin@domain-a.com，后续的API调用就作用于domain-a；换成admin@domain-b.com，就作用于domain-b。这意味着你的脚本只需要维护一个域名到管理员邮箱的映射表，就能实现跨域操作。

2.4 技术栈选择

调用管理API的方式有很多种，官方提供了Python、Java、Node.js、PHP等多种客户端库。我选Python，理由是生态成熟、脚本编写快、调试方便。核心依赖就两个：google-api-python-client和google-auth。前者封装了API调用，后者处理认证。不需要额外装什么重型框架，一个脚本文件加一个配置文件就能跑起来。

如果你更习惯用命令行工具，官方也有CLI工具可以调用部分API，但批量操作和复杂逻辑还是脚本更灵活。下面所有示例都以Python为准。

3. 核心配置环节的实操要点

3.1 服务账号的创建与密钥管理

创建服务账号的入口在云平台的控制台里，路径是"IAM与管理"→"服务账号"→"创建服务账号"。名称随便起，比如gadmin-multi-domain，描述写清楚用途。创建完成后，进入这个服务账号的详情页，找到"密钥"标签，点"添加密钥"→"创建新密钥"→选JSON格式，下载下来的文件就是你的凭证。

这个JSON文件里包含client_email、private_key等关键字段。client_email是服务账号的唯一标识，格式类似xxx@yyy.iam.gserviceaccount.com，后面配置DWD的时候要用到。private_key是私钥，用来签名请求。

实操心得：密钥文件下载后，建议立刻重命名为有意义的名字，比如sa-gadmin-prod.json，并且放到项目目录之外的统一凭证目录。我习惯在项目里放一个.env文件记录密钥路径，代码里通过环境变量读取，这样换环境的时候只改路径不改代码。

3.2 域级授权的配置细节

这一步是整个流程里最容易出错的环节。配置入口在管理控制台，路径是"安全"→"访问和数据控制"→"API控制"→"域级授权"。点"添加新"之后，填入服务账号的client_email，然后在"OAuth范围"里填入需要的Scope。

常用的Scope有这么几个：

• https://www.googleapis.com/auth/admin.directory.user：读写用户信息
• https://www.googleapis.com/auth/admin.directory.user.readonly：只读用户信息
• https://www.googleapis.com/auth/admin.directory.group：读写群组
• https://www.googleapis.com/auth/admin.directory.orgunit：读写组织单位

如果你不确定需要哪些，可以先给admin.directory.user这个最常用的。多个Scope之间用逗号分隔。

注意：Scope的URL必须一字不差，多一个斜杠少一个字母都会导致授权失败。我见过有人把admin.directory.user写成admin.directory.users，排查了半天才发现是多了个s。配置完成后，授权生效通常需要几分钟到几十分钟不等，别急着测试，等一会儿再试。

3.3 多域名的授权映射

如果你要管理多个域名，每个域名的管理控制台都需要单独配置一次DWD。也就是说，domain-a的管理员在domain-a的控制台里给这个服务账号授权，domain-b的管理员在domain-b的控制台里做同样的操作。服务账号本身是同一个，但每个域都要单独"认可"它。

这里有个前提：服务账号所属的项目和你要管理的域名之间不需要有从属关系。服务账号可以建在任意项目里，只要各个域的管理员愿意给它授权就行。这给了很大的灵活性——你可以用一个中立项目里的服务账号，去管理多个业务域名。

3.4 API的启用

在云平台的项目里，需要启用Admin SDK API。路径是"API和服务"→"库"，搜索"Admin SDK"，点进去启用。这一步不做的话，调用API会直接报"API未启用"的错误。启用之后等一两分钟再测试。

3.5 认证代码的编写

认证部分的核心逻辑是：加载服务账号密钥，用with_subject指定要模拟的管理员，然后构建API客户端。代码大概长这样：

from google.oauth2 import service_account from googleapiclient.discovery import build SCOPES = ['https://www.googleapis.com/auth/admin.directory.user'] def get_directory_service(key_file, admin_email): creds = service_account.Credentials.from_service_account_file( key_file, scopes=SCOPES) delegated = creds.with_subject(admin_email) return build('admin', 'directory_v1', credentials=delegated)

admin_email必须是目标域里拥有管理员权限的账号。如果这个账号没有管理员权限，API会返回403。with_subject这一步就是DWD的体现——服务账号"变成"了这个管理员。

实操心得：build函数的第三个参数可以加cache_discovery=False来避免一些缓存相关的警告。另外，如果你要频繁调用API，建议把service对象缓存起来复用，不要每次调用都重新build，那样会重复走认证流程，拖慢速度。

4. 跨域用户管理的完整实操流程

4.1 批量查询所有域名的用户列表

先从一个最简单的场景入手：拉取所有域名的用户清单。这个操作只需要读权限，风险最低，适合用来验证配置是否正确。

核心API是users().list()，关键参数有两个：domain指定要查询的域名，maxResults控制每页返回的数量（最大500）。如果用户数超过500，需要用pageToken翻页。

def list_all_users(service, domain): users = [] page_token = None while True: result = service.users().list( domain=domain, maxResults=500, pageToken=page_token, orderBy='email' ).execute() users.extend(result.get('users', [])) page_token = result.get('nextPageToken') if not page_token: break return users

翻页逻辑是必须的。我一开始图省事只取了第一页，结果一个有两千多人的域名只拉回来500条，差点漏掉关键数据。orderBy='email'能让结果按邮箱排序，方便后续对比。

跨域查询就是循环调用这个函数，每个域名传不同的admin_email：

domains = { 'domain-a.com': 'admin@domain-a.com', 'domain-b.com': 'admin@domain-b.com', } all_users = {} for domain, admin in domains.items(): svc = get_directory_service(KEY_FILE, admin) all_users[domain] = list_all_users(svc, domain)

跑完之后all_users就是一个字典，key是域名，value是该域的用户列表。你可以把它导出成CSV，或者直接做统计分析。

4.2 批量创建用户的参数详解

创建用户的API是users().insert()，请求体是一个包含用户信息的字典。必填字段有这几个：

• primaryEmail：主邮箱地址，必须属于目标域名
• name.givenName：名
• name.familyName：姓
• password：初始密码

可选字段里比较常用的有orgUnitPath（组织单位路径）、changePasswordAtNextLogin（下次登录是否强制改密）、suspended（是否停用）。

def create_user(service, email, given_name, family_name, password): user_body = { 'primaryEmail': email, 'name': { 'givenName': given_name, 'familyName': family_name, }, 'password': password, 'changePasswordAtNextLogin': True, } return service.users().insert(body=user_body).execute()

changePasswordAtNextLogin强烈建议设为True。你给的是初始密码，让用户第一次登录时自己改掉，这样初始密码即使泄露也不会有长期风险。

注意：密码必须满足域的密码策略要求。如果域配置了强密码策略（比如要求大小写加数字加特殊字符、最短12位），你传一个简单密码会直接报400错误。建议在脚本里加一个密码生成函数，自动生成符合策略的随机密码。

批量创建的时候，建议加一个try-except包裹每次调用，把失败的记录单独收集起来。因为批量操作里出现个别失败是常态——可能是邮箱已存在、可能是密码不合规、可能是组织单位路径写错了。如果不用异常处理，一个失败就会中断整个循环。

4.3 跨域用户同步的实现

用户同步是个更复杂的场景：把domain-a的用户信息同步到domain-b，或者从HR系统同步到所有域名。核心逻辑是"对比+差异操作"——先拉取两边现有的用户列表，对比出需要创建、更新、停用的账号，然后分别执行操作。

对比的关键字段是primaryEmail。如果domain-a有个用户zhangsan@domain-a.com，你要在domain-b创建对应的zhangsan@domain-b.com，那就需要做一个邮箱前缀的映射。

def sync_users(source_users, target_service, target_domain): target_users = {u['primaryEmail'] for u in list_all_users(target_service, target_domain)} created, skipped = [], [] for u in source_users: prefix = u['primaryEmail'].split('@')[0] target_email = f"{prefix}@{target_domain}" if target_email in target_users: skipped.append(target_email) continue try: create_user(target_service, target_email, u['name']['givenName'], u['name']['familyName'], generate_password()) created.append(target_email) except Exception as e: print(f"创建失败 {target_email}: {e}") return created, skipped

这个逻辑看起来简单，但实际跑的时候有几个坑。一是重名问题：如果domain-a有两个用户都叫"张伟"，邮箱前缀可能一个是zhangwei一个是zhangwei2，同步到domain-b的时候要保证不冲突。二是离职用户：源域里已经停用的账号，同步到目标域时应该也停用，而不是创建成活跃账号。三是组织单位映射：两个域的组织架构可能不一样，不能直接照搬orgUnitPath。

4.4 批量重置密码与强制下线

密码重置用users().update()，把password字段改成新值，同时把changePasswordAtNextLogin设为True。如果要强制用户下线（比如怀疑账号被盗），可以调用users().signOut()，这个操作会让该用户在所有设备上的会话失效。

def reset_password(service, email, new_password): body = { 'password': new_password, 'changePasswordAtNextLogin': True, } return service.users().update(userKey=email, body=body).execute() def force_signout(service, email): return service.users().signOut(userKey=email).execute()

userKey参数可以填邮箱地址，也可以填用户的唯一ID。用邮箱更直观，但如果用户改过邮箱，旧邮箱可能失效，这时候用ID更稳妥。

实操心得：批量重置密码之前，一定要先确认好新密码的分发方式。我见过有团队批量重置了所有人的密码，结果新密码只存在脚本的输出日志里，用户收不到通知，整个公司登录不了。正确做法是重置后通过备用邮箱或短信把新密码发给用户，或者干脆让用户走自助找回流程。

4.5 组织单位的批量管理

组织单位（Organizational Unit，简称OU）是管理用户的重要维度。你可以按部门、按地区、按项目建OU，然后把用户归到对应的OU里。批量管理OU的API是orgunits().insert()、orgunits().update()、orgunits().delete()。

创建OU的时候，parentOrgUnitPath指定父级路径，name是OU名称，orgUnitPath是完整路径。比如要在根目录下建一个"技术部"的OU：

def create_org_unit(service, name, parent_path='/'): body = { 'name': name, 'parentOrgUnitPath': parent_path, } return service.orgunits().insert(customerId='my_customer', body=body).execute()

customerId填my_customer就行，系统会自动识别当前域。如果要建多级OU，比如"技术部/后端组"，先建"技术部"，再以它为父级建"后端组"。

批量把用户移到指定OU，用users().update()改orgUnitPath字段：

def move_user_to_ou(service, email, ou_path): body = {'orgUnitPath': ou_path} return service.users().update(userKey=email, body=body).execute()

这个操作在组织架构调整的时候特别有用。比如公司重组，要把一批人从"产品部"移到"技术部"，手动在后台拖拽几十次，用脚本就是几行代码的事。

5. 常见报错与排查技巧实录

5.1 403错误的三类原因

403是这套方案里出现频率最高的错误，没有之一。根据我的经验，403基本可以归为三类：

第一类是DWD没配或配错。表现是报错信息里提到"unauthorized_client"或"access_denied"。排查方法是回到管理控制台，确认服务账号的client_email是否填对，Scope是否完整。特别注意Scope的URL不能有拼写错误。

第二类是subject账号没有管理员权限。表现是报错信息里提到"not authorized to access this resource"。排查方法是确认with_subject里填的邮箱确实是目标域的管理员，而且管理员权限没有被回收。

第三类是API没启用。表现是报错信息里提到"accessNotConfigured"或"API has not been used"。排查方法是去云平台项目里确认Admin SDK API已经启用。

5.2 404错误的定位方法

404通常意味着你操作的资源不存在。比如查询一个不存在的用户、更新一个已删除的OU。排查的时候，先用users().get()确认资源是否存在，再执行后续操作。

还有一种情况是域名写错了。比如你要操作domain-a.com，结果subject填的是admin@domain-b.com，API会在domain-b里找用户，自然找不到。这种错误很隐蔽，因为报错信息不会直接告诉你域名不对。

5.3 429限流与重试策略

API有调用频率限制，短时间内大量请求会触发429错误。批量操作几百上千个用户的时候，这个问题几乎必然出现。解决办法是加退避重试：遇到429就等一段时间再试，等待时间逐次递增。

import time from googleapiclient.errors import HttpError def call_with_retry(func, max_retries=5): for attempt in range(max_retries): try: return func() except HttpError as e: if e.resp.status == 429: wait = 2 ** attempt time.sleep(wait) else: raise raise Exception("重试次数耗尽")

退避策略用指数增长，第一次等1秒，第二次2秒，第三次4秒，以此类推。这样既能避免持续冲击API，又能在限流解除后尽快恢复。

实操心得：批量操作的时候，我习惯在每次API调用之间加一个0.1秒的间隔。虽然会拖慢整体速度，但能大幅降低触发限流的概率。对于几百个用户的操作，多花几十秒换来的稳定性是值得的。

5.4 密码策略冲突的处理

创建用户时如果密码不符合域的密码策略，会报400错误，信息里通常会说"Password does not meet the requirements"。解决办法是写一个密码生成函数，确保生成的密码满足常见策略要求：至少12位、包含大小写字母、包含数字、包含特殊字符。

import secrets import string def generate_password(length=16): alphabet = string.ascii_letters + string.digits + "!@#$%^&*" while True: pwd = ''.join(secrets.choice(alphabet) for _ in range(length)) if (any(c.islower() for c in pwd) and any(c.isupper() for c in pwd) and any(c.isdigit() for c in pwd) and any(c in "!@#$%^&*" for c in pwd)): return pwd

用secrets模块而不是random，因为secrets是密码学安全的，生成的密码更难预测。

5.5 多域名场景下的常见混淆

管理多个域名时，最容易搞混的就是subject和domain这两个参数。subject决定"以谁的身份操作"，domain决定"操作哪个域的数据"。如果subject是domain-a的管理员，但domain参数填了domain-b，API会报错，因为domain-a的管理员没有domain-b的权限。

我的做法是在脚本里维护一个配置字典，把域名和管理员邮箱绑定在一起，每次操作都从这个字典里取，避免手写参数时搞混：

DOMAIN_CONFIG = { 'domain-a.com': {'admin': 'admin@domain-a.com'}, 'domain-b.com': {'admin': 'admin@domain-b.com'}, }

这样调用的时候只需要传域名，管理员邮箱自动匹配，减少出错概率。

6. 几个提升效率的实战技巧

6.1 用配置文件管理多域参数

把域名、管理员邮箱、密钥路径这些参数全部抽到配置文件里，代码只负责逻辑。配置文件可以用YAML或JSON格式，我用YAML因为可读性好：

domains: domain-a.com: admin: admin@domain-a.com key_file: /path/to/sa-key.json domain-b.com: admin: admin@domain-b.com key_file: /path/to/sa-key.json

这样换环境的时候只改配置文件，代码一行不动。团队协作的时候，每个人用自己的配置文件，也不会互相干扰。

6.2 操作日志与审计追踪

批量操作一定要记日志。记什么？操作时间、操作类型、目标用户、执行结果。日志格式建议用结构化格式（比如JSON Lines），方便后续用工具分析。

import logging import json logger = logging.getLogger('gadmin') handler = logging.FileHandler('operations.log') handler.setFormatter(logging.Formatter('%(message)s')) logger.addHandler(handler) def log_operation(op_type, email, result): logger.info(json.dumps({ 'timestamp': time.time(), 'operation': op_type, 'email': email, 'result': result, }))

日志文件建议按天切割，保留至少90天。万一操作出了问题，日志是追溯原因的唯一依据。

6.3 先干跑再实操

批量操作之前，先用dry_run模式跑一遍。所谓干跑，就是把所有操作逻辑走一遍，但不真正调用API，只打印出"将要执行什么操作"。确认输出符合预期之后，再去掉dry_run标志正式执行。

def create_user_safe(service, email, given_name, family_name, password, dry_run=True): if dry_run: print(f"[DRY RUN] 将创建用户: {email}") return None return create_user(service, email, given_name, family_name, password)

这个习惯帮我避免过好几次事故。有一次我差点把一个域的所有用户都停用了，就是因为筛选条件写反了。干跑的时候看到输出是"将停用2000个用户"，立刻发现不对，及时改了过来。

6.4 分批处理大数量用户

如果一个域有几千甚至上万用户，一次性全拉回来内存扛不住，API也容易超时。解决办法是分批处理，每批处理500个，处理完一批再拉下一批。

def process_users_in_batches(service, domain, batch_size=500): page_token = None while True: result = service.users().list( domain=domain, maxResults=batch_size, pageToken=page_token ).execute() batch = result.get('users', []) yield batch page_token = result.get('nextPageToken') if not page_token: break

用生成器（generator）而不是一次性返回所有数据，内存占用小，处理起来也更灵活。

6.5 密钥轮换的平滑过渡

服务账号的密钥不是永久有效的，建议定期轮换。轮换的时候要注意平滑过渡：先创建新密钥，更新配置，确认新密钥能正常工作，再删除旧密钥。不要先删旧密钥再建新密钥，那样中间会有一段服务不可用的窗口。

轮换周期建议6到12个月。如果密钥有泄露风险，立即轮换。轮换之后，记得检查所有使用这个密钥的脚本和系统，确保它们都更新到了新密钥。

这套东西我用了挺长时间，从最开始管一个域，到后来管十几个域，脚本从几十行涨到几百行，但核心逻辑一直没变：服务账号加DWD加API调用。把这三个环节配好，剩下的就是业务逻辑的堆叠。真正花时间的从来不是写代码，而是搞清楚每个API的参数含义和边界条件。希望这篇东西能帮你少走点弯路，把时间花在更有价值的事情上。