即时访问授权（Just-In-Time Access，JITA）解决的是一个敏感问题：用户只在需要时获得权限，并在条件不再满足后失去权限。随着审批条件、资源类型和治理要求增加，把所有判断堆进一段条件代码，很快会形成难以解释、难以测试、也难以审计的授权逻辑。
HubSpot 对 JITA 授权系统的重构给出了一条清晰路径：把访问决策拆成独立规则，再用有向无环图（DAG）表达规则之间的依赖。同时，系统为决策增加结构化元数据、规则级可观测性和治理工作流。变化的重点不只是整理代码，而是让授权决策成为可以解释、测量和管理的工程对象。
从条件分支转向规则图
传统实现通常从一个入口函数开始，依次检查用户、资源、审批状态和风险信号：
if 用户有效 and 完成审批 and 资源允许JITA:
if 低风险 or 存在额外批准:
授权
这种写法在条件较少时足够直接，但随着业务发展会出现几个问题：
- 同一个条件被复制到多个授权流程中，修正规则时容易遗漏。
- 条件之间存在隐含顺序，开发者必须阅读完整函数才能发现依赖。
- 日志通常只记录最终的允许或拒绝，无法回答具体是哪条规则导致拒绝。
- 修改一项策略往往需要重新发布整个授权服务，治理边界不清晰。
规则引擎把每个判断封装为具有稳定输入、输出和身份标识的节点。例如，user_active 判断用户状态，approval_valid 验证审批，risk_acceptable 处理风险信号，grant_access 汇总上游结果。
DAG 则显式表示执行依赖：只有依赖节点完成后，当前规则才会执行。由于图中不存在环，系统可以进行拓扑排序，也能在启动或配置发布阶段拒绝循环依赖。对于彼此独立的规则，还可以并行求值以降低整体延迟。
决策结果不能只有 true 和 false
授权系统的调用方确实需要一个布尔结果，但运维、审计和治理流程需要更多上下文。一个更有用的规则结果可以包含：
{
"rule_id": "approval_valid",
"outcome": "deny",
"reason_code": "APPROVAL_EXPIRED",
"duration_ms": 3.7,
"policy_version": "2025-01",
"evidence": {
"approval_id": "apr-2048"
}
}
这类结构化元数据带来三项直接收益。
其一，系统能生成稳定的机器可读原因码。调用方不需要解析日志文本，就可以区分“审批过期”和“用户被停用”。
其二，可观测性可以下沉到规则级别。团队能够按 rule_id 统计延迟、错误率和拒绝率，而不是只监控整个授权接口。如果外部审批服务变慢，相关规则的耗时会直接暴露出来。
其三，审计记录可以关联策略版本、请求身份和证据来源，帮助团队回答“当时依据哪一版策略作出了什么决定”。不过，证据字段必须控制敏感信息，尤其不应把令牌、完整个人资料或机密资源内容写入日志。
一个可运行的最小规则 DAG
下面是一个可以直接运行的 Python 示例，用来演示独立规则、依赖检查和结构化决策轨迹。它是根据摘要所述架构编写的简化实践示例，并非 HubSpot 内部实现。
将代码保存为 jita_rules.py，然后执行 python jita_rules.py：
from dataclasses import asdict, dataclass
from time import perf_counter
from typing import Any, Callable
@dataclass(frozen=True)
class RuleResult:
rule_id: str
allowed: bool
reason_code: str
duration_ms: float
@dataclass(frozen=True)
class Rule:
rule_id: str
dependencies: tuple[str, ...]
evaluate: Callable[[dict[str, Any]], tuple[bool, str]]
def active_user(ctx: dict[str, Any]) -> tuple[bool, str]:
allowed = ctx.get("user_status") == "active"
return allowed, "USER_ACTIVE" if allowed else "USER_INACTIVE"
def valid_approval(ctx: dict[str, Any]) -> tuple[bool, str]:
allowed = bool(ctx.get("approval_valid"))
return allowed, "APPROVAL_VALID" if allowed else "APPROVAL_MISSING_OR_EXPIRED"
def acceptable_risk(ctx: dict[str, Any]) -> tuple[bool, str]:
allowed = ctx.get("risk_score", 100) < 50
return allowed, "RISK_ACCEPTABLE" if allowed else "RISK_TOO_HIGH"
def grant(_: dict[str, Any]) -> tuple[bool, str]:
return True, "ACCESS_GRANTED"
RULES = {
rule.rule_id: rule
for rule in (
Rule("user_active", (), active_user),
Rule("approval_valid", (), valid_approval),
Rule("risk_acceptable", ("user_active",), acceptable_risk),
Rule(
"grant_access",
("approval_valid", "risk_acceptable"),
grant,
),
)
}
def evaluate(target: str, ctx: dict[str, Any]) -> tuple[bool, list[RuleResult]]:
results: dict[str, RuleResult] = {}
visiting: set[str] = set()
def run(rule_id: str) -> RuleResult:
if rule_id in results:
return results[rule_id]
if rule_id in visiting:
raise ValueError(f"Cycle detected at rule: {rule_id}")
if rule_id not in RULES:
raise KeyError(f"Unknown rule: {rule_id}")
visiting.add(rule_id)
rule = RULES[rule_id]
dependency_results = [run(dep) for dep in rule.dependencies]
visiting.remove(rule_id)
denied = next((item for item in dependency_results if not item.allowed), None)
if denied:
result = RuleResult(
rule_id,
False,
f"DEPENDENCY_DENIED:{denied.rule_id}",
0.0,
)
else:
started = perf_counter()
allowed, reason = rule.evaluate(ctx)
duration_ms = (perf_counter() - started) * 1000
result = RuleResult(rule_id, allowed, reason, duration_ms)
results[rule_id] = result
return result
decision = run(target)
return decision.allowed, list(results.values())
if __name__ == "__main__":
request_context = {
"user_status": "active",
"approval_valid": True,
"risk_score": 18,
}
allowed, trace = evaluate("grant_access", request_context)
print({"allowed": allowed, "trace": [asdict(item) for item in trace]})
生产实现还需要补上超时、异常隔离、策略版本、并发控制和持久化审计。对外部依赖规则尤其要明确失败策略：审批服务超时时，是拒绝访问、使用短期缓存，还是进入人工复核。对高权限资源，一般应采用默认拒绝，而不是在依赖故障时放行。
让规则变更进入治理流程
规则引擎减少了条件代码的耦合，却也可能把复杂度转移到配置和流程中。如果任何人都能即时修改生产规则，系统只会从“代码难治理”变成“配置难治理”。
可以这样实践一套最小治理链路：
- 每条规则都有所有者、稳定 ID、说明和版本号。
- 规则变更通过代码审查或策略审查，不允许直接覆盖生产配置。
- 发布前验证 DAG 不含环、不引用未知节点，并覆盖关键允许与拒绝路径。
- 新版本先运行影子评估，只记录新旧结果差异，不影响实际授权。
- 按规则监控拒绝率、错误率、P95 延迟和新旧策略分歧率。
- 审计日志记录请求 ID、规则结果和策略版本，同时对敏感证据做脱敏和保留期限控制。
影子评估尤其适合授权系统。团队可以在真实流量上比较旧逻辑与新规则图，定位意外扩大权限或过度拒绝的问题，再决定是否切换执行版本。
采用时要守住的边界
规则 DAG 适合依赖关系明确、决策需要解释、策略持续演进的授权场景。它并不会自动解决所有权限问题：规则数量失控会形成庞大的图；跨规则共享可变状态会破坏独立性；过度记录证据又会产生新的隐私风险。
落地时应优先挑选一条条件复杂但边界清楚的 JITA 流程，先拆出稳定规则和原因码，再接入规则级指标与审计。等团队能够可靠地测试、发布和回滚规则后，再扩大覆盖范围。衡量重构是否成功，也不应只看代码行数，而要看一次拒绝能否被快速解释、一次策略变更能否被安全验证，以及一次异常能否精确定位到具体规则。