-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathjson_contracts.py
More file actions
156 lines (127 loc) · 4.67 KB
/
Copy pathjson_contracts.py
File metadata and controls
156 lines (127 loc) · 4.67 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
"""Versioned JSON contracts for machine-facing CLI consumers.
The helpers in this module deliberately return plain dictionaries so a
consumer can extend ``details`` without depending on a framework-specific
model class. Field names and their meanings are part of the public v1
contract and are documented in ``docs/json-contracts.md``.
"""
from __future__ import annotations
import json
import logging
import re
from collections.abc import Mapping
from datetime import datetime, timezone
from logging import LogRecord
from typing import Any
from .redaction import REDACTED, redact_text_value
JSON_CONTRACT_VERSION = 1
JSON_LOG_SCHEMA = "base-cli.log"
JSON_OUTPUT_SCHEMA = "base-cli.output"
JSON_ERROR_SCHEMA = "base-cli.error"
MAX_JSON_LOG_MESSAGE_LENGTH = 8192
_SENSITIVE_ASSIGNMENT_BOUNDARY = (
r"(?=(?:[&,;]\s*(?=[A-Za-z][A-Za-z0-9_-]*\s*[=:])"
r"|\s+[A-Za-z][A-Za-z0-9_-]*\s*[=:])|\s|$)"
)
_SENSITIVE_ASSIGNMENT = re.compile(
r"(?i)(\b(?:token|password|secret|api[-_]?key|authorization)\b\s*[:=]\s*)"
rf"(\S+?){_SENSITIVE_ASSIGNMENT_BOUNDARY}"
)
__all__ = [
"JSON_CONTRACT_VERSION",
"JSON_ERROR_SCHEMA",
"JSON_LOG_SCHEMA",
"JSON_OUTPUT_SCHEMA",
"JsonLogFormatter",
"MAX_JSON_LOG_MESSAGE_LENGTH",
"error_envelope",
"success_envelope",
"dumps_envelope",
"redact_json_value",
]
def success_envelope(
*,
run_id: str | None,
details: Mapping[str, Any] | None = None,
message: str = "Success",
code: str = "ok",
) -> dict[str, Any]:
"""Return the stable v1 machine-readable success envelope."""
return {
"schema_version": JSON_CONTRACT_VERSION,
"schema": JSON_OUTPUT_SCHEMA,
"code": code,
"type": "success",
"message": _safe_text(message),
"details": redact_json_value(dict(details or {})),
"run_id": run_id,
}
def error_envelope(
*,
run_id: str | None,
code: str,
message: str,
details: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
"""Return the stable v1 machine-readable error envelope."""
return {
"schema_version": JSON_CONTRACT_VERSION,
"schema": JSON_ERROR_SCHEMA,
"code": _safe_text(code),
"type": "error",
"message": _safe_text(message),
"details": redact_json_value(dict(details or {})),
"run_id": run_id,
}
def dumps_envelope(envelope: Mapping[str, Any]) -> str:
"""Serialize an envelope as one compact, newline-terminated JSON record."""
return (
json.dumps(
redact_json_value(dict(envelope)),
ensure_ascii=False,
separators=(",", ":"),
)
+ "\n"
)
def redact_json_value(value: Any, *, _key: str | None = None) -> Any:
"""Recursively redact secret-looking JSON keys and text values."""
if _key is not None and _is_sensitive_key(_key):
return REDACTED
if isinstance(value, Mapping):
return {str(key): redact_json_value(item, _key=str(key)) for key, item in value.items()}
if isinstance(value, list):
return [redact_json_value(item) for item in value]
if isinstance(value, tuple):
return [redact_json_value(item) for item in value]
if isinstance(value, str):
return _safe_text(value)
return value
class JsonLogFormatter(logging.Formatter):
"""Format one ``LogRecord`` as a bounded, redacted JSON object."""
def __init__(self, run_id: str | None = None) -> None:
super().__init__()
self.run_id = run_id
def format(self, record: LogRecord) -> str:
message = _safe_text(record.getMessage())
if len(message) > MAX_JSON_LOG_MESSAGE_LENGTH:
message = message[:MAX_JSON_LOG_MESSAGE_LENGTH] + "…"
payload: dict[str, Any] = {
"schema_version": JSON_CONTRACT_VERSION,
"schema": JSON_LOG_SCHEMA,
"timestamp": _timestamp(record.created),
"level": record.levelname,
"logger": record.name,
"message": message,
"run_id": self.run_id,
}
if record.exc_info and record.exc_info[0] is not None:
payload["details"] = {
"exception_type": record.exc_info[0].__name__,
}
return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
def _timestamp(value: float) -> str:
return datetime.fromtimestamp(value, tz=timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
def _safe_text(value: str) -> str:
redacted = redact_text_value(value)
return _SENSITIVE_ASSIGNMENT.sub(r"\1" + REDACTED, redacted)
def _is_sensitive_key(value: str) -> bool:
return re.search(r"(?i)(token|password|secret|api[-_]?key|authorization)", value) is not None