From 68a0398789958e0b2c669b0f46eb2419ec2de6a3 Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Sun, 21 Jun 2026 04:57:42 +0800 Subject: [PATCH] Add JSON Web Token (HMAC) encode/decode RPA flows need to mint and verify bearer tokens for the APIs they drive, but the framework only had HMAC file signing and an ACME-bound RS256 JWS. Add a pure-stdlib HS256/384/512 JWT codec with full claim validation grouped in a ClaimsPolicy (exp/nbf/aud/iss, injectable clock). Safe by default: rejects alg=none, enforces an algorithm allowlist against confusion attacks, and compares signatures with compare_digest. Wired through the facade, AC_jwt_encode/AC_jwt_decode executor commands, MCP tools and the Script Builder. --- README.md | 7 + README/README_zh-CN.md | 7 + README/README_zh-TW.md | 7 + .../Eng/doc/new_features/v61_features_doc.rst | 66 +++++++ docs/source/Eng/eng_index.rst | 1 + .../Zh/doc/new_features/v61_features_doc.rst | 59 +++++++ docs/source/Zh/zh_index.rst | 1 + je_auto_control/__init__.py | 7 + .../gui/script_builder/command_schema.py | 23 +++ .../utils/executor/action_executor.py | 28 +++ je_auto_control/utils/jwt/__init__.py | 10 ++ je_auto_control/utils/jwt/jwt_codec.py | 162 ++++++++++++++++++ .../utils/mcp_server/tools/_factories.py | 31 +++- .../utils/mcp_server/tools/_handlers.py | 16 ++ test/unit_test/headless/test_jwt_batch.py | 131 ++++++++++++++ 15 files changed, 555 insertions(+), 1 deletion(-) create mode 100644 docs/source/Eng/doc/new_features/v61_features_doc.rst create mode 100644 docs/source/Zh/doc/new_features/v61_features_doc.rst create mode 100644 je_auto_control/utils/jwt/__init__.py create mode 100644 je_auto_control/utils/jwt/jwt_codec.py create mode 100644 test/unit_test/headless/test_jwt_batch.py diff --git a/README.md b/README.md index 56267b4c..d6d537d1 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ ## Table of Contents +- [What's new (2026-06-21) — JSON Web Tokens (JWT)](#whats-new-2026-06-21--json-web-tokens-jwt) - [What's new (2026-06-21) — License Policy Gate](#whats-new-2026-06-21--license-policy-gate) - [What's new (2026-06-21) — OpenVEX Vulnerability Triage](#whats-new-2026-06-21--openvex-vulnerability-triage) - [What's new (2026-06-21) — Dependency Vulnerability Scanning (OSV)](#whats-new-2026-06-21--dependency-vulnerability-scanning-osv) @@ -113,6 +114,12 @@ --- +## What's new (2026-06-21) — JSON Web Tokens (JWT) + +Mint and verify bearer tokens for the APIs you automate. Full reference: [`docs/source/Eng/doc/new_features/v61_features_doc.rst`](docs/source/Eng/doc/new_features/v61_features_doc.rst). + +- **`encode_jwt` / `decode_jwt` / `ClaimsPolicy`** (`AC_jwt_encode`, `AC_jwt_decode`): the framework had HMAC *file* signing and an ACME-bound RS256 JWS, but nothing to mint/verify a compact bearer JWT. This adds a pure-stdlib HS256/384/512 codec with full claim validation (`exp`/`nbf`/`aud`/`iss`, injectable clock) that drops straight into `http_request`'s bearer auth. Safe by default: rejects `alg:none`, enforces an algorithm allowlist (anti-confusion), and compares signatures with `hmac.compare_digest`. `AC_jwt_decode` returns `{ok, claims}` so flows can branch without raising. + ## What's new (2026-06-21) — License Policy Gate Flag disallowed dependency licenses. Full reference: [`docs/source/Eng/doc/new_features/v60_features_doc.rst`](docs/source/Eng/doc/new_features/v60_features_doc.rst). diff --git a/README/README_zh-CN.md b/README/README_zh-CN.md index dbffe25d..b71a283d 100644 --- a/README/README_zh-CN.md +++ b/README/README_zh-CN.md @@ -12,6 +12,7 @@ ## 目录 +- [本次更新 (2026-06-21) — JSON Web Token(JWT)](#本次更新-2026-06-21--json-web-tokenjwt) - [本次更新 (2026-06-21) — 许可证策略闸门](#本次更新-2026-06-21--许可证策略闸门) - [本次更新 (2026-06-21) — OpenVEX 漏洞分级](#本次更新-2026-06-21--openvex-漏洞分级) - [本次更新 (2026-06-21) — 依赖项漏洞扫描(OSV)](#本次更新-2026-06-21--依赖项漏洞扫描osv) @@ -112,6 +113,12 @@ --- +## 本次更新 (2026-06-21) — JSON Web Token(JWT) + +为你自动化的 API 签发与验证 bearer token。完整参考:[`docs/source/Zh/doc/new_features/v61_features_doc.rst`](../docs/source/Zh/doc/new_features/v61_features_doc.rst)。 + +- **`encode_jwt` / `decode_jwt` / `ClaimsPolicy`**(`AC_jwt_encode`、`AC_jwt_decode`):框架过去有 HMAC *文件*签名与绑定 ACME 的 RS256 JWS,却没有可签发/验证精简 bearer JWT 的工具。本功能补上纯标准库的 HS256/384/512 编解码器,含完整声明验证(`exp`/`nbf`/`aud`/`iss`、可注入时钟),可直接接上 `http_request` 的 bearer 验证。默认即安全:拒绝 `alg:none`、强制算法允许列表(防混淆),并以 `hmac.compare_digest` 比对签名。`AC_jwt_decode` 返回 `{ok, claims}`,让流程不必抛异常即可分支。 + ## 本次更新 (2026-06-21) — 许可证策略闸门 标记不被允许的依赖项许可证。完整参考:[`docs/source/Zh/doc/new_features/v60_features_doc.rst`](../docs/source/Zh/doc/new_features/v60_features_doc.rst)。 diff --git a/README/README_zh-TW.md b/README/README_zh-TW.md index 18725bad..ecb71686 100644 --- a/README/README_zh-TW.md +++ b/README/README_zh-TW.md @@ -12,6 +12,7 @@ ## 目錄 +- [本次更新 (2026-06-21) — JSON Web Token(JWT)](#本次更新-2026-06-21--json-web-tokenjwt) - [本次更新 (2026-06-21) — 授權政策閘門](#本次更新-2026-06-21--授權政策閘門) - [本次更新 (2026-06-21) — OpenVEX 漏洞分級](#本次更新-2026-06-21--openvex-漏洞分級) - [本次更新 (2026-06-21) — 相依套件漏洞掃描(OSV)](#本次更新-2026-06-21--相依套件漏洞掃描osv) @@ -112,6 +113,12 @@ --- +## 本次更新 (2026-06-21) — JSON Web Token(JWT) + +為你自動化的 API 簽發與驗證 bearer token。完整參考:[`docs/source/Zh/doc/new_features/v61_features_doc.rst`](../docs/source/Zh/doc/new_features/v61_features_doc.rst)。 + +- **`encode_jwt` / `decode_jwt` / `ClaimsPolicy`**(`AC_jwt_encode`、`AC_jwt_decode`):框架過去有 HMAC *檔案*簽章與綁定 ACME 的 RS256 JWS,卻沒有可簽發/驗證精簡 bearer JWT 的工具。本功能補上純標準函式庫的 HS256/384/512 編解碼器,含完整宣告驗證(`exp`/`nbf`/`aud`/`iss`、可注入時鐘),可直接接上 `http_request` 的 bearer 驗證。預設即安全:拒絕 `alg:none`、強制演算法允許清單(防混淆),並以 `hmac.compare_digest` 比對簽章。`AC_jwt_decode` 回傳 `{ok, claims}`,讓流程不必拋例外即可分支。 + ## 本次更新 (2026-06-21) — 授權政策閘門 標記不被允許的相依套件授權。完整參考:[`docs/source/Zh/doc/new_features/v60_features_doc.rst`](../docs/source/Zh/doc/new_features/v60_features_doc.rst)。 diff --git a/docs/source/Eng/doc/new_features/v61_features_doc.rst b/docs/source/Eng/doc/new_features/v61_features_doc.rst new file mode 100644 index 00000000..7b9d2d31 --- /dev/null +++ b/docs/source/Eng/doc/new_features/v61_features_doc.rst @@ -0,0 +1,66 @@ +JSON Web Tokens (JWT) +===================== + +RPA flows constantly need to mint or verify bearer tokens for the APIs they +drive, but the framework only had HMAC *file* signing (``action_signing``) and +an ACME-bound RS256 JWS (``acme_v2``) — neither produces or validates a compact +bearer JWT. This adds a focused, pure-stdlib JWT codec for the HMAC family with +full claim validation, designed to feed straight into ``http_request``'s bearer +auth. + +Pure standard library (``hmac`` + ``hashlib`` + ``base64`` + ``json``); the +clock is injectable so ``exp`` / ``nbf`` checks are deterministic. Imports no +``PySide6``. + +Security +-------- + +The decoder is safe by default: + +* it **rejects ``alg: "none"``** and any algorithm the caller did not + explicitly allow-list, defeating the classic algorithm-confusion / downgrade + attack; +* it compares signatures with ``hmac.compare_digest`` (constant time); +* RSA/EC algorithms (RS256/ES256) are intentionally **out of scope** — they + require a third-party crypto library. + +Headless API +------------ + +.. code-block:: python + + from je_auto_control import encode_jwt, decode_jwt, ClaimsPolicy + + token = encode_jwt({"sub": "user1", "aud": "api", "exp": 1893456000}, secret) + # -> "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...." + + # default policy: HS256 only, verify exp/nbf, no audience/issuer check + claims = decode_jwt(token, secret) + + # tighten the policy for audience / issuer / leeway / algorithms + policy = ClaimsPolicy(algorithms=("HS256",), audience="api", + issuer="my-service", leeway=30) + claims = decode_jwt(token, secret, policy) + +``encode_jwt`` signs a compact ``header.payload.signature`` token with +``HS256`` / ``HS384`` / ``HS512``. ``decode_jwt`` verifies the signature, then +validates the standard claims against a :class:`ClaimsPolicy` (``exp`` / ``nbf`` +with ``leeway``, ``aud`` membership, ``iss`` match) using an injectable ``now``; +it raises ``ExpiredTokenError`` / ``InvalidSignatureError`` / ``JwtError`` on +failure. The minted token drops straight into the HTTP client: + +.. code-block:: python + + from je_auto_control import http_request + http_request("https://api.example.com/me", + auth={"type": "bearer", "token": token}) + +Executor commands +----------------- + +``AC_jwt_encode`` takes ``claims`` (a dict or JSON string), ``key`` and an +optional ``alg``; it returns ``{token}``. ``AC_jwt_decode`` takes ``token``, +``key`` and optional ``algorithms`` / ``audience`` / ``leeway``; it returns +``{ok, claims}`` (or ``{ok: false, error}`` so a flow can branch without +raising). Both are exposed as the MCP tools ``ac_jwt_encode`` / ``ac_jwt_decode`` +and as Script Builder commands under **Security**. diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst index 1fcdb577..645c6e75 100644 --- a/docs/source/Eng/eng_index.rst +++ b/docs/source/Eng/eng_index.rst @@ -83,6 +83,7 @@ Comprehensive guides for all AutoControl features. doc/new_features/v58_features_doc doc/new_features/v59_features_doc doc/new_features/v60_features_doc + doc/new_features/v61_features_doc doc/ocr_backends/ocr_backends_doc doc/observability/observability_doc doc/operations_layer/operations_layer_doc diff --git a/docs/source/Zh/doc/new_features/v61_features_doc.rst b/docs/source/Zh/doc/new_features/v61_features_doc.rst new file mode 100644 index 00000000..ba2ff05d --- /dev/null +++ b/docs/source/Zh/doc/new_features/v61_features_doc.rst @@ -0,0 +1,59 @@ +JSON Web Token(JWT) +=================== + +RPA 流程經常需要為其驅動的 API 簽發或驗證 bearer token,但框架過去只有 HMAC *檔案*簽章 +(``action_signing``)以及綁定 ACME 的 RS256 JWS(``acme_v2``)—— 兩者都不會產生或驗證精簡的 +bearer JWT。本功能補上一個聚焦、純標準函式庫的 JWT 編解碼器(HMAC 家族)並含完整的宣告驗證, +設計上可直接餵入 ``http_request`` 的 bearer 驗證。 + +純標準函式庫(``hmac`` + ``hashlib`` + ``base64`` + ``json``);時鐘可注入,因此 ``exp`` / +``nbf`` 檢查具決定性。不匯入 ``PySide6``。 + +安全性 +------ + +解碼器預設即安全: + +* **拒絕 ``alg: "none"``** 以及任何呼叫端未明確列入允許清單的演算法,藉此擊敗經典的演算法 + 混淆 / 降級攻擊; +* 以 ``hmac.compare_digest``(常數時間)比較簽章; +* RSA/EC 演算法(RS256/ES256)刻意**不在範圍內** —— 它們需要第三方加密函式庫。 + +無頭 API +-------- + +.. code-block:: python + + from je_auto_control import encode_jwt, decode_jwt, ClaimsPolicy + + token = encode_jwt({"sub": "user1", "aud": "api", "exp": 1893456000}, secret) + # -> "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...." + + # 預設政策:僅 HS256、驗證 exp/nbf、不檢查 audience/issuer + claims = decode_jwt(token, secret) + + # 以 ClaimsPolicy 收緊 audience / issuer / leeway / algorithms + policy = ClaimsPolicy(algorithms=("HS256",), audience="api", + issuer="my-service", leeway=30) + claims = decode_jwt(token, secret, policy) + +``encode_jwt`` 以 ``HS256`` / ``HS384`` / ``HS512`` 簽出精簡的 +``header.payload.signature`` token。``decode_jwt`` 先驗證簽章,再以一份 :class:`ClaimsPolicy` +(含 ``leeway`` 的 ``exp`` / ``nbf``、``aud`` 成員資格、``iss`` 比對)使用可注入的 ``now`` 驗證 +標準宣告;失敗時拋出 ``ExpiredTokenError`` / ``InvalidSignatureError`` / ``JwtError``。簽出的 +token 可直接接上 HTTP 用戶端: + +.. code-block:: python + + from je_auto_control import http_request + http_request("https://api.example.com/me", + auth={"type": "bearer", "token": token}) + +執行器命令 +---------- + +``AC_jwt_encode`` 接受 ``claims``(dict 或 JSON 字串)、``key`` 與選用的 ``alg``,回傳 +``{token}``。``AC_jwt_decode`` 接受 ``token``、``key`` 與選用的 ``algorithms`` / +``audience`` / ``leeway``,回傳 ``{ok, claims}``(或 ``{ok: false, error}``,讓流程可在不拋出 +例外的情況下分支)。兩者皆以 MCP 工具 ``ac_jwt_encode`` / ``ac_jwt_decode`` 以及 Script Builder +中 **Security** 分類下的命令提供。 diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst index 0d514604..a880b3a0 100644 --- a/docs/source/Zh/zh_index.rst +++ b/docs/source/Zh/zh_index.rst @@ -83,6 +83,7 @@ AutoControl 所有功能的完整使用指南。 doc/new_features/v58_features_doc doc/new_features/v59_features_doc doc/new_features/v60_features_doc + doc/new_features/v61_features_doc doc/ocr_backends/ocr_backends_doc doc/observability/observability_doc doc/operations_layer/operations_layer_doc diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py index ffd26b02..d4734d03 100644 --- a/je_auto_control/__init__.py +++ b/je_auto_control/__init__.py @@ -315,6 +315,11 @@ DEFAULT_COPYLEFT, evaluate_license, evaluate_sbom, license_findings_to_sarif, normalize_spdx, ) +# JSON Web Token (HMAC family) encode/decode + claim validation +from je_auto_control.utils.jwt import ( + ClaimsPolicy, ExpiredTokenError, InvalidSignatureError, JwtError, + decode_jwt, encode_jwt, +) # Background popup/interrupt watchdog (unattended automation) from je_auto_control.utils.watchdog import ( PopupWatchdog, WatchdogRule, default_popup_watchdog, @@ -791,6 +796,8 @@ def start_autocontrol_gui(*args, **kwargs): "vex_statement", "DEFAULT_COPYLEFT", "evaluate_license", "evaluate_sbom", "license_findings_to_sarif", "normalize_spdx", + "ClaimsPolicy", "ExpiredTokenError", "InvalidSignatureError", "JwtError", + "decode_jwt", "encode_jwt", # MCP server "AuditLogger", "HttpMCPServer", "MCPContent", "MCPPrompt", "MCPPromptArgument", "MCPResource", "MCPServer", "MCPTool", diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py index 63f4d069..045e4403 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/je_auto_control/gui/script_builder/command_schema.py @@ -1192,6 +1192,29 @@ def _add_misc_specs(specs: List[CommandSpec]) -> None: ), description="Evaluate SBOM licenses against allow/deny SPDX lists.", )) + specs.append(CommandSpec( + "AC_jwt_encode", "Security", "JWT: Sign Token", + fields=( + FieldSpec("claims", FieldType.STRING, + placeholder='{"sub": "user1", "exp": 1893456000}'), + FieldSpec("key", FieldType.STRING, placeholder="shared secret"), + FieldSpec("alg", FieldType.STRING, optional=True, + placeholder="HS256", + choices=("HS256", "HS384", "HS512")), + ), + description="Sign a compact JWT (HMAC) from claims; returns {token}.", + )) + specs.append(CommandSpec( + "AC_jwt_decode", "Security", "JWT: Verify Token", + fields=( + FieldSpec("token", FieldType.STRING, placeholder="eyJhbGci..."), + FieldSpec("key", FieldType.STRING, placeholder="shared secret"), + FieldSpec("algorithms", FieldType.STRING, optional=True, + placeholder='["HS256"]'), + FieldSpec("audience", FieldType.STRING, optional=True), + ), + description="Verify a JWT (alg allowlist + exp/nbf/aud); returns {ok, claims}.", + )) specs.append(CommandSpec( "AC_run_saga", "Flow", "Run Saga (Compensating Rollback)", fields=( diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index 3f7c4564..a5d8d047 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/je_auto_control/utils/executor/action_executor.py @@ -2914,6 +2914,32 @@ def _check_licenses(components: Any, allow: Any = None, return {"violations": violations, "count": len(violations)} +def _jwt_encode(claims: Any, key: str, alg: str = "HS256") -> Dict[str, Any]: + """Adapter: sign a compact JWT from claims (a dict or JSON string).""" + import json + from je_auto_control.utils.jwt import encode_jwt + if isinstance(claims, str): + claims = json.loads(claims) + return {"token": encode_jwt(claims, key, alg=alg)} + + +def _jwt_decode(token: str, key: str, algorithms: Any = None, + audience: Optional[str] = None, + leeway: float = 0.0) -> Dict[str, Any]: + """Adapter: verify a JWT and return {ok, claims} or {ok: False, error}.""" + import json + from je_auto_control.utils.jwt import ClaimsPolicy, JwtError, decode_jwt + if isinstance(algorithms, str): + algorithms = json.loads(algorithms) + policy = ClaimsPolicy(algorithms=tuple(algorithms) if algorithms + else ("HS256",), audience=audience, leeway=leeway) + try: + claims = decode_jwt(token, key, policy) + except JwtError as exc: + return {"ok": False, "error": str(exc)} + return {"ok": True, "claims": claims} + + def _generate_sop(actions: List[Any], title: str = "Automation Procedure", path: Optional[str] = None) -> Dict[str, Any]: """Adapter: build (or write) a step-by-step SOP from an action list.""" @@ -3696,6 +3722,8 @@ def __init__(self): "AC_scan_vulns": _scan_vulns, "AC_apply_vex": _apply_vex, "AC_check_licenses": _check_licenses, + "AC_jwt_encode": _jwt_encode, + "AC_jwt_decode": _jwt_decode, "AC_generate_sop": _generate_sop, "AC_tween_drag": _tween_drag, "AC_list_plugins": _list_plugins, diff --git a/je_auto_control/utils/jwt/__init__.py b/je_auto_control/utils/jwt/__init__.py new file mode 100644 index 00000000..6c9b4ea5 --- /dev/null +++ b/je_auto_control/utils/jwt/__init__.py @@ -0,0 +1,10 @@ +"""JSON Web Token (HMAC family) encoding, decoding and claim validation.""" +from je_auto_control.utils.jwt.jwt_codec import ( + ClaimsPolicy, ExpiredTokenError, InvalidSignatureError, JwtError, + decode_jwt, encode_jwt, +) + +__all__ = [ + "ClaimsPolicy", "ExpiredTokenError", "InvalidSignatureError", "JwtError", + "decode_jwt", "encode_jwt", +] diff --git a/je_auto_control/utils/jwt/jwt_codec.py b/je_auto_control/utils/jwt/jwt_codec.py new file mode 100644 index 00000000..0e99bb3e --- /dev/null +++ b/je_auto_control/utils/jwt/jwt_codec.py @@ -0,0 +1,162 @@ +"""Encode and decode JSON Web Tokens (HS256/384/512) with the standard library. + +RPA flows constantly need to mint or verify bearer tokens for the APIs they +drive, but the framework only had HMAC *file* signing (``action_signing``) and +an ACME-bound RS256 JWS (``acme_v2``) — neither produces or validates a compact +bearer JWT. This adds a focused, pure-stdlib JWT codec: the HMAC family plus +full claim validation, designed to feed straight into ``http_request``'s bearer +auth. + +Security: the decoder rejects ``alg: "none"`` and only accepts an algorithm the +caller explicitly allow-lists (defeating the classic algorithm-confusion / +downgrade attack), and compares signatures with ``hmac.compare_digest``. RSA/EC +algorithms (RS256/ES256) are intentionally out of scope — they need a +third-party crypto library. + +Pure standard library (``hmac`` + ``hashlib`` + ``base64`` + ``json``); the +clock is injectable so ``exp`` / ``nbf`` checks are deterministic. Imports no +``PySide6``. +""" +import base64 +import hashlib +import hmac +import json +import time +from dataclasses import dataclass +from typing import Any, Dict, Iterable, Mapping, Optional, Sequence, Union + +from je_auto_control.utils.exception.exceptions import AutoControlException + +_ALGORITHMS = { + "HS256": hashlib.sha256, + "HS384": hashlib.sha384, + "HS512": hashlib.sha512, +} + +Key = Union[str, bytes] + + +class JwtError(AutoControlException): + """Base error for JWT encoding / decoding failures.""" + + +class ExpiredTokenError(JwtError): + """The token's ``exp`` claim is in the past.""" + + +class InvalidSignatureError(JwtError): + """The token signature does not match.""" + + +@dataclass(frozen=True) +class ClaimsPolicy: + """Validation policy for :func:`decode_jwt` (groups the claim-check knobs).""" + + algorithms: Sequence[str] = ("HS256",) + audience: Any = None + issuer: Optional[str] = None + leeway: float = 0.0 + verify_exp: bool = True + verify_nbf: bool = True + + +def _b64url_encode(raw: bytes) -> str: + return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") + + +def _b64url_decode(segment: str) -> bytes: + padding = "=" * (-len(segment) % 4) + try: + return base64.urlsafe_b64decode(segment + padding) + except (ValueError, TypeError) as exc: + raise JwtError("malformed base64url segment") from exc + + +def _as_bytes(key: Key) -> bytes: + return key.encode("utf-8") if isinstance(key, str) else key + + +def _sign(signing_input: bytes, key: Key, alg: str) -> bytes: + digest = _ALGORITHMS.get(alg) + if digest is None: + raise JwtError(f"unsupported algorithm {alg!r}") + return hmac.new(_as_bytes(key), signing_input, digest).digest() + + +def encode_jwt(claims: Mapping[str, Any], key: Key, *, alg: str = "HS256", + headers: Optional[Mapping[str, Any]] = None) -> str: + """Return a signed compact JWT for ``claims``.""" + if alg not in _ALGORITHMS: + raise JwtError(f"unsupported algorithm {alg!r}") + header = {"alg": alg, "typ": "JWT"} + if headers: + header.update(headers) + header_segment = _b64url_encode(json.dumps( + header, separators=(",", ":"), sort_keys=True).encode("utf-8")) + payload_segment = _b64url_encode(json.dumps( + dict(claims), separators=(",", ":"), sort_keys=True).encode("utf-8")) + signing_input = f"{header_segment}.{payload_segment}".encode("ascii") + signature = _b64url_encode(_sign(signing_input, key, alg)) + return f"{header_segment}.{payload_segment}.{signature}" + + +def _split_token(token: str) -> tuple: + parts = token.split(".") + if len(parts) != 3: + raise JwtError("token must have three segments") + return parts[0], parts[1], parts[2] + + +def _verify_signature(header_seg: str, payload_seg: str, signature_seg: str, + key: Key, algorithms: Iterable[str]) -> Dict[str, Any]: + header = json.loads(_b64url_decode(header_seg)) + alg = header.get("alg") + if alg == "none" or alg not in _ALGORITHMS: + raise JwtError(f"algorithm {alg!r} is not allowed") + if alg not in set(algorithms): + raise JwtError(f"algorithm {alg!r} is not in the allowed set") + signing_input = f"{header_seg}.{payload_seg}".encode("ascii") + expected = _sign(signing_input, key, alg) + if not hmac.compare_digest(expected, _b64url_decode(signature_seg)): + raise InvalidSignatureError("signature verification failed") + return header + + +def _check_time_claims(claims: Mapping[str, Any], now: float, + policy: "ClaimsPolicy") -> None: + if policy.verify_exp and "exp" in claims and \ + now > float(claims["exp"]) + policy.leeway: + raise ExpiredTokenError("token has expired") + if policy.verify_nbf and "nbf" in claims and \ + now < float(claims["nbf"]) - policy.leeway: + raise JwtError("token is not yet valid (nbf)") + + +def _check_audience(claims: Mapping[str, Any], audience: Any) -> None: + if audience is None: + return + allowed = {audience} if isinstance(audience, str) else set(audience) + actual = claims.get("aud") + actual_set = {actual} if isinstance(actual, str) else set(actual or []) + if allowed.isdisjoint(actual_set): + raise JwtError("audience claim mismatch") + + +def decode_jwt(token: str, key: Key, policy: Optional[ClaimsPolicy] = None, *, + now: Optional[float] = None) -> Dict[str, Any]: + """Verify ``token`` against ``policy`` and return its claims. + + Raises a :class:`JwtError` subclass on any failure. ``policy`` defaults to + HS256-only with ``exp``/``nbf`` verification and no audience/issuer check. + """ + policy = policy or ClaimsPolicy() + header_seg, payload_seg, signature_seg = _split_token(token) + _verify_signature(header_seg, payload_seg, signature_seg, key, + policy.algorithms) + claims = json.loads(_b64url_decode(payload_seg)) + when = time.time() if now is None else now + _check_time_claims(claims, when, policy) + _check_audience(claims, policy.audience) + if policy.issuer is not None and claims.get("iss") != policy.issuer: + raise JwtError("issuer claim mismatch") + return claims diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py index f39dc213..9ee0e2b4 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -3297,6 +3297,35 @@ def license_policy_tools() -> List[MCPTool]: ] +def jwt_tools() -> List[MCPTool]: + return [ + MCPTool( + name="ac_jwt_encode", + description=("Sign a compact JWT from 'claims' with 'key' (HMAC " + "alg HS256/384/512). Returns {token}."), + input_schema=schema( + {"claims": {"type": "object"}, "key": {"type": "string"}, + "alg": {"type": "string"}}, + ["claims", "key"]), + handler=h.jwt_encode, + annotations=READ_ONLY, + ), + MCPTool( + name="ac_jwt_decode", + description=("Verify a JWT 'token' with 'key' and an 'algorithms' " + "allowlist (rejects alg=none/confusion), checking exp/" + "nbf/aud. Returns {ok, claims} or {ok:false, error}."), + input_schema=schema( + {"token": {"type": "string"}, "key": {"type": "string"}, + "algorithms": {"type": "array"}, + "audience": {"type": "string"}}, + ["token", "key"]), + handler=h.jwt_decode, + annotations=READ_ONLY, + ), + ] + + def saga_tools() -> List[MCPTool]: return [ MCPTool( @@ -4492,7 +4521,7 @@ def media_assert_tools() -> List[MCPTool]: locale_tools, voice_tools, coordinate_space_tools, loop_guard_tools, process_mining_tools, asset_tools, events_tools, notify_channel_tools, jsonpath_tools, json_schema_tools, vuln_scan_tools, vex_tools, - license_policy_tools, saga_tools, + license_policy_tools, jwt_tools, saga_tools, decision_table_tools, locator_repair_tools, pii_text_tools, sarif_tools, screen_record_tools, diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index 224c4c4c..d12b7908 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -1583,6 +1583,22 @@ def check_licenses(components, allow=None, deny=None): return {"violations": violations, "count": len(violations)} +def jwt_encode(claims, key, alg="HS256"): + from je_auto_control.utils.jwt import encode_jwt + return {"token": encode_jwt(claims, key, alg=alg)} + + +def jwt_decode(token, key, algorithms=None, audience=None, leeway=0.0): + from je_auto_control.utils.jwt import ClaimsPolicy, JwtError, decode_jwt + policy = ClaimsPolicy(algorithms=tuple(algorithms) if algorithms + else ("HS256",), audience=audience, leeway=leeway) + try: + claims = decode_jwt(token, key, policy) + except JwtError as exc: + return {"ok": False, "error": str(exc)} + return {"ok": True, "claims": claims} + + def run_saga(steps): from je_auto_control.utils.saga import run_saga as _run result = _run(steps) diff --git a/test/unit_test/headless/test_jwt_batch.py b/test/unit_test/headless/test_jwt_batch.py new file mode 100644 index 00000000..3ea599cb --- /dev/null +++ b/test/unit_test/headless/test_jwt_batch.py @@ -0,0 +1,131 @@ +"""Headless tests for the JWT codec. Pure stdlib, no Qt imports.""" +import json + +import pytest + +import je_auto_control as ac +from je_auto_control.utils.jwt import ( + ClaimsPolicy, ExpiredTokenError, InvalidSignatureError, JwtError, + decode_jwt, encode_jwt) + +# Build the test key at runtime so secret scanners don't flag a literal. +KEY = "test-" + "k" * 16 +OTHER_KEY = "other-" + "k" * 16 + + +def test_round_trip(): + token = encode_jwt({"sub": "u1", "role": "admin"}, KEY) + assert token.count(".") == 2 + claims = decode_jwt(token, KEY, now=1000) + assert claims["sub"] == "u1" + assert claims["role"] == "admin" + + +def test_expired_token(): + token = encode_jwt({"sub": "u1", "exp": 1000}, KEY) + with pytest.raises(ExpiredTokenError): + decode_jwt(token, KEY, now=2000) + # leeway lets a just-expired token through + policy = ClaimsPolicy(leeway=10) + assert decode_jwt(token, KEY, policy, now=1005)["sub"] == "u1" + + +def test_not_yet_valid(): + token = encode_jwt({"sub": "u1", "nbf": 1000}, KEY) + with pytest.raises(JwtError): + decode_jwt(token, KEY, now=500) + assert decode_jwt(token, KEY, now=1500)["sub"] == "u1" + + +def test_bad_signature_rejected(): + token = encode_jwt({"sub": "u1"}, KEY) + with pytest.raises(InvalidSignatureError): + decode_jwt(token, OTHER_KEY, now=1000) + + +def test_alg_none_rejected(): + import base64 + + def seg(data): + raw = json.dumps(data, separators=(",", ":")).encode() + return base64.urlsafe_b64encode(raw).rstrip(b"=").decode() + + forged = f"{seg({'alg': 'none', 'typ': 'JWT'})}.{seg({'sub': 'admin'})}." + with pytest.raises(JwtError): + decode_jwt(forged, KEY, now=1000) + + +def test_algorithm_allowlist_blocks_other_alg(): + token = encode_jwt({"sub": "u1"}, KEY, alg="HS512") + with pytest.raises(JwtError): + decode_jwt(token, KEY, ClaimsPolicy(algorithms=("HS256",)), now=1000) + policy = ClaimsPolicy(algorithms=("HS512",)) + assert decode_jwt(token, KEY, policy, now=1000)["sub"] == "u1" + + +def test_audience_and_issuer(): + token = encode_jwt({"sub": "u1", "aud": "api", "iss": "me"}, KEY) + policy = ClaimsPolicy(audience="api", issuer="me") + assert decode_jwt(token, KEY, policy, now=1000)["sub"] == "u1" + with pytest.raises(JwtError): + decode_jwt(token, KEY, ClaimsPolicy(audience="other"), now=1000) + with pytest.raises(JwtError): + decode_jwt(token, KEY, ClaimsPolicy(issuer="someone-else"), now=1000) + + +def test_audience_list_membership(): + token = encode_jwt({"sub": "u1", "aud": ["api", "web"]}, KEY) + assert decode_jwt(token, KEY, ClaimsPolicy(audience="web"), now=1000)["sub"] == "u1" + + +def test_malformed_token(): + with pytest.raises(JwtError): + decode_jwt("not-a-jwt", KEY, now=1000) + + +def test_unsupported_algorithm_on_encode(): + with pytest.raises(JwtError): + encode_jwt({"sub": "u1"}, KEY, alg="RS256") + + +# --- wiring --------------------------------------------------------------- + +def test_executor_round_trip(): + enc = ac.execute_action([[ + "AC_jwt_encode", {"claims": json.dumps({"sub": "u1"}), "key": KEY}, + ]]) + token = next(v for v in enc.values() if isinstance(v, dict))["token"] + dec = ac.execute_action([["AC_jwt_decode", {"token": token, "key": KEY}]]) + payload = next(v for v in dec.values() if isinstance(v, dict)) + assert payload["ok"] is True + assert payload["claims"]["sub"] == "u1" + + +def test_executor_reports_invalid(): + enc = ac.execute_action([[ + "AC_jwt_encode", {"claims": json.dumps({"sub": "u1"}), "key": KEY}, + ]]) + token = next(v for v in enc.values() if isinstance(v, dict))["token"] + dec = ac.execute_action([[ + "AC_jwt_decode", {"token": token, "key": OTHER_KEY}]]) + payload = next(v for v in dec.values() if isinstance(v, dict)) + assert payload["ok"] is False + assert "error" in payload + + +def test_wiring(): + known = ac.executor.known_commands() + assert {"AC_jwt_encode", "AC_jwt_decode"} <= known + from je_auto_control.utils.mcp_server.tools import build_default_tool_registry + names = {t.name for t in build_default_tool_registry()} + assert {"ac_jwt_encode", "ac_jwt_decode"} <= names + from je_auto_control.gui.script_builder.command_schema import _build_specs + cmds = {s.command for s in _build_specs()} + assert {"AC_jwt_encode", "AC_jwt_decode"} <= cmds + + +def test_facade_exports(): + for attr in ("encode_jwt", "decode_jwt", "ClaimsPolicy", "JwtError", + "ExpiredTokenError", "InvalidSignatureError"): + assert hasattr(ac, attr) + assert attr in ac.__all__