b003394568
dev와 dev-corp 요청을 명시적 환경 프로필로 라우팅하고, private override 없이 단일 실행기에서 Edge 검증과 환경별 metadata 정책을 처리한다.
1363 lines
47 KiB
Python
1363 lines
47 KiB
Python
"""Single-entry dev and dev-corp IOP token issuance workflow.
|
|
|
|
Sensitive request data is accepted only as JSON on stdin. The command emits a
|
|
redacted result and copies the raw token to the local macOS clipboard only after
|
|
Edge activation, environment-specific API smoke, metadata synchronization, and
|
|
leak checks pass.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import fcntl
|
|
import hashlib
|
|
import html
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import shlex
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import unicodedata
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from contextlib import contextmanager
|
|
from html.parser import HTMLParser
|
|
from pathlib import Path
|
|
from typing import Any, NoReturn, cast
|
|
|
|
ATLASSIAN_BASE_URL = "https://lgucorp.atlassian.net"
|
|
CONFLUENCE_FOLDER_ID = "650886407"
|
|
CONFLUENCE_TITLE = "IOP 계정 발급 현황"
|
|
MANAGED_HEADING = "IOP 사용자 토큰 발급 현황"
|
|
TABLE_HEADERS = ("사용자", "principal alias", "token ref", "상태", "동기화 시각")
|
|
SUPPORTED_ENVIRONMENTS = ("dev", "dev-corp")
|
|
|
|
|
|
class WorkflowFailure(RuntimeError):
|
|
def __init__(self, code: str):
|
|
self.code = code
|
|
super().__init__(code)
|
|
|
|
|
|
class ConfluenceHTTPFailure(WorkflowFailure):
|
|
def __init__(self, status_code: int):
|
|
self.status_code = status_code
|
|
super().__init__(f"confluence_http_{status_code}")
|
|
|
|
|
|
class CurlTransportFailure(WorkflowFailure):
|
|
def __init__(self):
|
|
super().__init__("curl_transport_failed")
|
|
|
|
|
|
def fail(code: str) -> NoReturn:
|
|
raise WorkflowFailure(code)
|
|
|
|
|
|
def repo_root() -> Path:
|
|
root = Path(__file__).resolve().parents[5]
|
|
if not (root / ".git").exists():
|
|
fail("repo_root_invalid")
|
|
return root
|
|
|
|
|
|
def load_profile(root: Path, environment: str) -> dict[str, Any]:
|
|
if environment not in SUPPORTED_ENVIRONMENTS:
|
|
fail("environment_not_supported")
|
|
path = (
|
|
root
|
|
/ "agent-ops/skills/project/openai-usage-token-issue/profiles"
|
|
/ f"{environment}.json"
|
|
)
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
fail("environment_profile_invalid")
|
|
if not isinstance(value, dict) or value.get("environment") != environment:
|
|
fail("environment_profile_invalid")
|
|
required_strings = (
|
|
"edge_ssh",
|
|
"secret_store",
|
|
"openai_smoke_transport",
|
|
"openai_base_url",
|
|
"smoke_model",
|
|
"metrics_transport",
|
|
"metrics_url",
|
|
)
|
|
if any(not isinstance(value.get(key), str) or not value[key] for key in required_strings):
|
|
fail("environment_profile_invalid")
|
|
if not isinstance(value.get("confluence_enabled"), bool):
|
|
fail("environment_profile_invalid")
|
|
expected_stores = {
|
|
"dev": "token/.dev-iop-token",
|
|
"dev-corp": "token/.dev-corp-iop-token",
|
|
}
|
|
expected_ssh = {"dev": "toki@toki-labs.com", "dev-corp": "toki@iop.ai.kr"}
|
|
if value["secret_store"] != expected_stores[environment]:
|
|
fail("environment_profile_invalid")
|
|
if value["edge_ssh"] != expected_ssh[environment]:
|
|
fail("environment_profile_invalid")
|
|
parsed_api = urllib.parse.urlsplit(str(value["openai_base_url"]))
|
|
transport = value["openai_smoke_transport"]
|
|
if transport == "public-https":
|
|
if parsed_api.scheme != "https" or not parsed_api.netloc:
|
|
fail("environment_profile_invalid")
|
|
elif transport == "ssh-loopback":
|
|
if parsed_api.scheme != "http" or parsed_api.hostname not in {"127.0.0.1", "::1", "localhost"}:
|
|
fail("environment_profile_invalid")
|
|
else:
|
|
fail("environment_profile_invalid")
|
|
if value["metrics_transport"] not in {"public-http", "ssh-loopback"}:
|
|
fail("environment_profile_invalid")
|
|
if value["confluence_enabled"] and not isinstance(
|
|
value.get("confluence_user_store"), str
|
|
):
|
|
fail("environment_profile_invalid")
|
|
return cast(dict[str, Any], value)
|
|
|
|
|
|
def run_quiet(
|
|
argv: list[str], *, cwd: Path, input_bytes: bytes | None = None, timeout: int = 30
|
|
) -> subprocess.CompletedProcess[bytes]:
|
|
try:
|
|
return subprocess.run(
|
|
argv,
|
|
cwd=cwd,
|
|
input=input_bytes,
|
|
capture_output=True,
|
|
timeout=timeout,
|
|
check=False,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
fail("command_timeout")
|
|
|
|
|
|
def git_is_ignored(root: Path, path: Path) -> bool:
|
|
relative = path.relative_to(root)
|
|
result = run_quiet(["git", "check-ignore", "-q", "--", str(relative)], cwd=root)
|
|
return result.returncode == 0
|
|
|
|
|
|
def git_is_tracked(root: Path, path: Path) -> bool:
|
|
relative = path.relative_to(root)
|
|
result = run_quiet(
|
|
["git", "ls-files", "--error-unmatch", "--", str(relative)], cwd=root
|
|
)
|
|
return result.returncode == 0
|
|
|
|
|
|
def require_secure_secret(root: Path, path: Path, *, must_exist: bool = True) -> None:
|
|
if path.is_symlink() or path.parent.is_symlink():
|
|
fail("secret_store_symlink_invalid")
|
|
try:
|
|
path.parent.resolve().relative_to(root.resolve())
|
|
except ValueError:
|
|
fail("secret_store_path_invalid")
|
|
if must_exist and not path.is_file():
|
|
fail("secret_store_missing")
|
|
if path.exists():
|
|
mode = stat.S_IMODE(path.stat().st_mode)
|
|
if mode != 0o600:
|
|
fail("secret_store_mode_invalid")
|
|
if not git_is_ignored(root, path) or git_is_tracked(root, path):
|
|
fail("secret_store_tracking_invalid")
|
|
|
|
|
|
@contextmanager
|
|
def local_transaction_lock(root: Path, *, shared: bool):
|
|
token_dir = root / "token"
|
|
if (
|
|
not token_dir.is_dir()
|
|
or token_dir.is_symlink()
|
|
or not git_is_ignored(root, token_dir)
|
|
):
|
|
fail("local_transaction_lock_invalid")
|
|
flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
|
|
try:
|
|
descriptor = os.open(token_dir, flags)
|
|
except OSError:
|
|
fail("local_transaction_lock_invalid")
|
|
try:
|
|
operation = fcntl.LOCK_SH if shared else fcntl.LOCK_EX
|
|
try:
|
|
fcntl.flock(descriptor, operation | fcntl.LOCK_NB)
|
|
except BlockingIOError:
|
|
fail("local_transaction_busy")
|
|
yield
|
|
finally:
|
|
try:
|
|
fcntl.flock(descriptor, fcntl.LOCK_UN)
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
|
|
def normalize_alias(principal_ref: str, requested: str | None) -> str:
|
|
source = requested.strip() if requested else principal_ref.split("@", 1)[0]
|
|
ascii_value = (
|
|
unicodedata.normalize("NFKD", source).encode("ascii", "ignore").decode("ascii")
|
|
)
|
|
alias = re.sub(r"[^a-z0-9]+", "-", ascii_value.lower()).strip("-")
|
|
alias = re.sub(r"-+", "-", alias)[:40].rstrip("-")
|
|
if not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,39}", alias):
|
|
fail("principal_alias_invalid")
|
|
return alias
|
|
|
|
|
|
def parse_request(stdin_text: str) -> dict[str, str]:
|
|
try:
|
|
payload = json.loads(stdin_text)
|
|
except json.JSONDecodeError:
|
|
fail("request_json_invalid")
|
|
if not isinstance(payload, dict):
|
|
fail("request_json_invalid")
|
|
if payload.get("operation", "create") != "create":
|
|
fail("operation_not_supported_by_single_entry")
|
|
environment = payload.get("env")
|
|
if not isinstance(environment, str) or environment not in SUPPORTED_ENVIRONMENTS:
|
|
fail("environment_not_supported")
|
|
principal_value = payload.get("principal_ref")
|
|
alias_value = payload.get("principal_alias")
|
|
token_ref_value = payload.get("token_ref")
|
|
if not isinstance(principal_value, str):
|
|
fail("principal_ref_invalid")
|
|
if alias_value is not None and not isinstance(alias_value, str):
|
|
fail("principal_alias_invalid")
|
|
if token_ref_value is not None and not isinstance(token_ref_value, str):
|
|
fail("token_ref_invalid")
|
|
principal_ref = principal_value.strip()
|
|
if (
|
|
not principal_ref
|
|
or len(principal_ref) > 254
|
|
or any(char in principal_ref for char in "\r\n:\0")
|
|
or any(char.isspace() for char in principal_ref)
|
|
):
|
|
fail("principal_ref_invalid")
|
|
alias = normalize_alias(principal_ref, alias_value)
|
|
token_ref = (token_ref_value or f"iop-{environment}-{alias}").strip()
|
|
if not re.fullmatch(r"[a-z0-9][a-z0-9._:-]{2,79}", token_ref):
|
|
fail("token_ref_invalid")
|
|
return {
|
|
"operation": "create",
|
|
"env": environment,
|
|
"principal_ref": principal_ref,
|
|
"principal_alias": alias,
|
|
"token_ref": token_ref,
|
|
}
|
|
|
|
|
|
def read_store(path: Path) -> tuple[bytes, dict[str, str]]:
|
|
if not path.exists():
|
|
return b"", {}
|
|
raw_bytes = path.read_bytes()
|
|
try:
|
|
text = raw_bytes.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
fail("secret_store_encoding_invalid")
|
|
entries: dict[str, str] = {}
|
|
for line in text.splitlines():
|
|
if not line.strip():
|
|
continue
|
|
key, separator, token = line.partition(": ")
|
|
if (
|
|
not separator
|
|
or not key
|
|
or key.strip() != key
|
|
or not token.startswith("iop_")
|
|
or token.strip() != token
|
|
or key in entries
|
|
or "\n" in token
|
|
or "\r" in token
|
|
):
|
|
fail("secret_store_format_invalid")
|
|
entries[key] = token
|
|
return raw_bytes, entries
|
|
|
|
|
|
def write_candidate(path: Path, content: bytes) -> Path:
|
|
path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
os.chmod(path.parent, 0o700)
|
|
candidate = (
|
|
path.parent / f".{path.name}.candidate-{os.getpid()}-{secrets.token_hex(4)}"
|
|
)
|
|
descriptor = os.open(candidate, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
try:
|
|
with os.fdopen(descriptor, "wb") as stream:
|
|
stream.write(content)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
except BaseException:
|
|
candidate.unlink(missing_ok=True)
|
|
raise
|
|
return candidate
|
|
|
|
|
|
def atomic_restore(path: Path, content: bytes | None) -> None:
|
|
if content is None:
|
|
path.unlink(missing_ok=True)
|
|
return
|
|
candidate = write_candidate(path, content)
|
|
os.replace(candidate, path)
|
|
os.chmod(path, 0o600)
|
|
|
|
|
|
def sha256_text(value: str) -> str:
|
|
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def curl_config_quote(value: str) -> str:
|
|
return (
|
|
value.replace("\\", "\\\\")
|
|
.replace('"', '\\"')
|
|
.replace("\r", "")
|
|
.replace("\n", "")
|
|
)
|
|
|
|
|
|
def curl_request(
|
|
root: Path,
|
|
method: str,
|
|
url: str,
|
|
headers: list[str],
|
|
payload: dict[str, Any] | None,
|
|
*,
|
|
timeout: int,
|
|
) -> tuple[int, bytes]:
|
|
marker = b"\n__IOP_HTTP_STATUS__"
|
|
config_lines = [
|
|
"silent",
|
|
"show-error",
|
|
'proto = "=https"',
|
|
f"max-time = {timeout}",
|
|
f'request = "{curl_config_quote(method)}"',
|
|
f'url = "{curl_config_quote(url)}"',
|
|
'write-out = "\\n__IOP_HTTP_STATUS__%{http_code}"',
|
|
]
|
|
config_lines.extend(f'header = "{curl_config_quote(header)}"' for header in headers)
|
|
argv = ["curl", "-q", "--config", "-"]
|
|
payload_path: Path | None = None
|
|
if payload is not None:
|
|
payload_bytes = json.dumps(
|
|
payload, ensure_ascii=False, separators=(",", ":")
|
|
).encode("utf-8")
|
|
payload_path = write_candidate(root / "token/.curl-json-payload", payload_bytes)
|
|
argv.extend(["--data-binary", f"@{payload_path}"])
|
|
try:
|
|
result = run_quiet(
|
|
argv,
|
|
cwd=root,
|
|
input_bytes=("\n".join(config_lines) + "\n").encode("utf-8"),
|
|
timeout=timeout + 5,
|
|
)
|
|
finally:
|
|
if payload_path:
|
|
payload_path.unlink(missing_ok=True)
|
|
if result.returncode != 0:
|
|
raise CurlTransportFailure()
|
|
body, separator, status_bytes = result.stdout.rpartition(marker)
|
|
if not separator or not status_bytes.isdigit():
|
|
raise CurlTransportFailure()
|
|
return int(status_bytes), body
|
|
|
|
|
|
def remote_call(
|
|
root: Path,
|
|
profile: dict[str, Any],
|
|
payload: dict[str, Any],
|
|
*,
|
|
timeout: int,
|
|
) -> dict[str, Any]:
|
|
helper = (
|
|
root
|
|
/ "agent-ops/skills/project/openai-usage-token-issue/scripts/remote_edge_transaction.rb"
|
|
)
|
|
source = helper.read_text(encoding="utf-8")
|
|
remote_command = "ruby -e " + shlex.quote(source)
|
|
remote_payload = dict(payload)
|
|
remote_payload["environment"] = profile["environment"]
|
|
result = run_quiet(
|
|
[
|
|
"ssh",
|
|
"-o",
|
|
"BatchMode=yes",
|
|
"-o",
|
|
"ConnectTimeout=8",
|
|
str(profile["edge_ssh"]),
|
|
remote_command,
|
|
],
|
|
cwd=root,
|
|
input_bytes=json.dumps(remote_payload, separators=(",", ":")).encode("utf-8"),
|
|
timeout=timeout,
|
|
)
|
|
parsed: dict[str, Any] | None = None
|
|
for line in reversed(result.stdout.decode("utf-8", "replace").splitlines()):
|
|
try:
|
|
value = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if isinstance(value, dict) and "status" in value:
|
|
parsed = value
|
|
break
|
|
if parsed is None:
|
|
fail("edge_remote_response_invalid")
|
|
if result.returncode != 0 or parsed.get("status") == "blocked":
|
|
code = str(parsed.get("code", "edge_remote_failed"))
|
|
if not re.fullmatch(r"[a-z0-9_]+", code):
|
|
code = "edge_remote_failed"
|
|
fail(code if code.startswith("edge_") else f"edge_{code}")
|
|
return parsed
|
|
|
|
|
|
class ConfluenceClient:
|
|
def __init__(self, root: Path, username: str, token: str):
|
|
self.root = root
|
|
self.username = username
|
|
self.token = token
|
|
|
|
def request(
|
|
self, method: str, path_or_url: str, payload: dict[str, Any] | None = None
|
|
) -> dict[str, Any]:
|
|
url = (
|
|
path_or_url
|
|
if path_or_url.startswith("https://")
|
|
else ATLASSIAN_BASE_URL + path_or_url
|
|
)
|
|
parsed_url = urllib.parse.urlsplit(url)
|
|
if (
|
|
parsed_url.scheme != "https"
|
|
or parsed_url.netloc != "lgucorp.atlassian.net"
|
|
or not parsed_url.path.startswith("/wiki/")
|
|
):
|
|
fail("confluence_url_invalid")
|
|
try:
|
|
status_code, data = curl_request(
|
|
self.root,
|
|
method,
|
|
url,
|
|
[
|
|
f"Authorization: Basic {base64_basic(self.username, self.token)}",
|
|
"Accept: application/json",
|
|
"Content-Type: application/json",
|
|
],
|
|
payload,
|
|
timeout=20,
|
|
)
|
|
except CurlTransportFailure:
|
|
fail("confluence_network_failed")
|
|
if status_code < 200 or status_code >= 300:
|
|
raise ConfluenceHTTPFailure(status_code)
|
|
try:
|
|
parsed = json.loads(data)
|
|
except json.JSONDecodeError:
|
|
fail("confluence_response_invalid")
|
|
if not isinstance(parsed, dict):
|
|
fail("confluence_response_invalid")
|
|
return parsed
|
|
|
|
|
|
def base64_basic(username: str, token: str) -> str:
|
|
return base64.b64encode(f"{username}:{token}".encode()).decode("ascii")
|
|
|
|
|
|
def confluence_storage(page: dict[str, Any]) -> str:
|
|
body = page.get("body")
|
|
if not isinstance(body, dict):
|
|
fail("confluence_body_missing")
|
|
storage = body.get("storage")
|
|
if not isinstance(storage, dict) or not isinstance(storage.get("value"), str):
|
|
fail("confluence_body_missing")
|
|
return str(storage["value"])
|
|
|
|
|
|
def page_version(page: dict[str, Any]) -> int:
|
|
version = page.get("version")
|
|
if not isinstance(version, dict) or not isinstance(version.get("number"), int):
|
|
fail("confluence_version_invalid")
|
|
return int(version["number"])
|
|
|
|
|
|
def strip_storage_tags(value: str) -> str:
|
|
without_tags = re.sub(r"<[^>]+>", "", value)
|
|
return re.sub(r"\s+", " ", html.unescape(without_tags)).strip()
|
|
|
|
|
|
def managed_section_bounds(storage_body: str) -> tuple[int, int]:
|
|
heading_pattern = re.compile(
|
|
r"<h([1-6])\b[^>]*>.*?</h\1>", re.IGNORECASE | re.DOTALL
|
|
)
|
|
headings = list(heading_pattern.finditer(storage_body))
|
|
matches = [
|
|
item
|
|
for item in headings
|
|
if strip_storage_tags(item.group(0)) == MANAGED_HEADING
|
|
]
|
|
if len(matches) != 1:
|
|
fail("confluence_managed_heading_invalid")
|
|
current = matches[0]
|
|
current_level = int(current.group(1))
|
|
end = len(storage_body)
|
|
for candidate in headings:
|
|
if (
|
|
candidate.start() > current.start()
|
|
and int(candidate.group(1)) <= current_level
|
|
):
|
|
end = candidate.start()
|
|
break
|
|
return current.start(), end
|
|
|
|
|
|
def managed_section(mapping_rows: list[dict[str, str]], synchronized_at: str) -> str:
|
|
parts = [f"<h2>{html.escape(MANAGED_HEADING)}</h2>", "<table><tbody><tr>"]
|
|
parts.extend(f"<th>{html.escape(header)}</th>" for header in TABLE_HEADERS)
|
|
parts.append("</tr>")
|
|
for item in sorted(mapping_rows, key=lambda row: row["principal_ref"]):
|
|
values = (
|
|
item["principal_ref"],
|
|
item["principal_alias"],
|
|
item["token_ref"],
|
|
"active",
|
|
synchronized_at,
|
|
)
|
|
parts.append("<tr>")
|
|
parts.extend(f"<td>{html.escape(value)}</td>" for value in values)
|
|
parts.append("</tr>")
|
|
parts.append("</tbody></table>")
|
|
return "".join(parts)
|
|
|
|
|
|
def replace_managed_section(storage_body: str, replacement: str) -> str:
|
|
if not storage_body.strip():
|
|
return replacement
|
|
start, end = managed_section_bounds(storage_body)
|
|
return storage_body[:start] + replacement + storage_body[end:]
|
|
|
|
|
|
class TableParser(HTMLParser):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.rows: list[list[str]] = []
|
|
self.current_row: list[str] | None = None
|
|
self.current_cell: list[str] | None = None
|
|
|
|
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
|
del attrs
|
|
if tag.lower() == "tr":
|
|
self.current_row = []
|
|
elif tag.lower() in {"td", "th"} and self.current_row is not None:
|
|
self.current_cell = []
|
|
|
|
def handle_data(self, data: str) -> None:
|
|
if self.current_cell is not None:
|
|
self.current_cell.append(data)
|
|
|
|
def handle_endtag(self, tag: str) -> None:
|
|
lowered = tag.lower()
|
|
if (
|
|
lowered in {"td", "th"}
|
|
and self.current_cell is not None
|
|
and self.current_row is not None
|
|
):
|
|
self.current_row.append(
|
|
re.sub(r"\s+", " ", "".join(self.current_cell)).strip()
|
|
)
|
|
self.current_cell = None
|
|
elif lowered == "tr" and self.current_row is not None:
|
|
self.rows.append(self.current_row)
|
|
self.current_row = None
|
|
|
|
|
|
def parse_managed_table(storage_body: str) -> list[tuple[str, str, str, str, str]]:
|
|
start, end = managed_section_bounds(storage_body)
|
|
section = storage_body[start:end]
|
|
if "iop_" in section or re.search(
|
|
r"(?<![0-9a-f])[0-9a-f]{64}(?![0-9a-f])", section, re.IGNORECASE
|
|
):
|
|
fail("confluence_secret_material_detected")
|
|
parser = TableParser()
|
|
parser.feed(section)
|
|
if not parser.rows or tuple(parser.rows[0]) != TABLE_HEADERS:
|
|
fail("confluence_table_header_invalid")
|
|
records: list[tuple[str, str, str, str, str]] = []
|
|
principals: set[str] = set()
|
|
for row in parser.rows[1:]:
|
|
if len(row) != 5 or not row[0] or not row[4]:
|
|
fail("confluence_table_row_invalid")
|
|
record = cast(tuple[str, str, str, str, str], tuple(row))
|
|
if record[0] in principals:
|
|
fail("confluence_table_principal_duplicate")
|
|
principals.add(record[0])
|
|
records.append(record)
|
|
return records
|
|
|
|
|
|
def verify_managed_table(storage_body: str, mapping_rows: list[dict[str, str]]) -> None:
|
|
records = parse_managed_table(storage_body)
|
|
expected = {
|
|
(item["principal_ref"], item["principal_alias"], item["token_ref"], "active")
|
|
for item in mapping_rows
|
|
}
|
|
actual = {record[:4] for record in records}
|
|
if len(records) != len(mapping_rows) or actual != expected:
|
|
fail("confluence_table_mismatch")
|
|
|
|
|
|
def ensure_create_preserves_rows(
|
|
existing_records: list[tuple[str, str, str, str, str]],
|
|
mapping_rows: list[dict[str, str]],
|
|
) -> None:
|
|
expected_principals = {item["principal_ref"] for item in mapping_rows}
|
|
existing_principals = {record[0] for record in existing_records}
|
|
if not existing_principals.issubset(expected_principals):
|
|
fail("confluence_create_would_remove_user")
|
|
|
|
|
|
def page_in_folder(client: ConfluenceClient, page: dict[str, Any]) -> bool:
|
|
parent_id = str(page.get("parentId") or "")
|
|
parent_type = str(page.get("parentType") or "")
|
|
seen: set[tuple[str, str]] = set()
|
|
for _ in range(20):
|
|
if parent_id == CONFLUENCE_FOLDER_ID:
|
|
return True
|
|
if not parent_id or (parent_type, parent_id) in seen:
|
|
return False
|
|
seen.add((parent_type, parent_id))
|
|
if parent_type == "page":
|
|
parent = client.request(
|
|
"GET", f"/wiki/api/v2/pages/{urllib.parse.quote(parent_id)}"
|
|
)
|
|
else:
|
|
parent = client.request(
|
|
"GET", f"/wiki/api/v2/folders/{urllib.parse.quote(parent_id)}"
|
|
)
|
|
parent_id = str(parent.get("parentId") or "")
|
|
parent_type = str(parent.get("parentType") or "")
|
|
return False
|
|
|
|
|
|
def exact_title_pages(client: ConfluenceClient, space_id: str) -> list[dict[str, Any]]:
|
|
query = urllib.parse.urlencode(
|
|
{
|
|
"space-id": space_id,
|
|
"title": CONFLUENCE_TITLE,
|
|
"status": "current",
|
|
"limit": 250,
|
|
}
|
|
)
|
|
next_url: str | None = f"/wiki/api/v2/pages?{query}"
|
|
matches: list[dict[str, Any]] = []
|
|
while next_url:
|
|
response = client.request("GET", next_url)
|
|
results = response.get("results")
|
|
if not isinstance(results, list):
|
|
fail("confluence_search_invalid")
|
|
for result in results:
|
|
if (
|
|
isinstance(result, dict)
|
|
and result.get("title") == CONFLUENCE_TITLE
|
|
and result.get("status") == "current"
|
|
):
|
|
matches.append(result)
|
|
links = response.get("_links")
|
|
next_value = links.get("next") if isinstance(links, dict) else None
|
|
next_url = str(next_value) if next_value else None
|
|
return matches
|
|
|
|
|
|
def preflight_confluence(
|
|
root: Path,
|
|
client: ConfluenceClient,
|
|
target_cache: Path,
|
|
*,
|
|
read_only: bool,
|
|
) -> dict[str, Any]:
|
|
client.request("GET", "/wiki/rest/api/user/current")
|
|
folder = client.request("GET", f"/wiki/api/v2/folders/{CONFLUENCE_FOLDER_ID}")
|
|
space_id = str(folder.get("spaceId") or "")
|
|
if not space_id:
|
|
fail("confluence_space_invalid")
|
|
candidates = exact_title_pages(client, space_id)
|
|
valid: list[dict[str, Any]] = []
|
|
for page_candidate in candidates:
|
|
page_id = str(page_candidate.get("id") or "")
|
|
if not page_id:
|
|
continue
|
|
detail = client.request(
|
|
"GET",
|
|
f"/wiki/api/v2/pages/{urllib.parse.quote(page_id)}?body-format=storage",
|
|
)
|
|
if page_in_folder(client, detail):
|
|
valid.append(detail)
|
|
if len(valid) > 1:
|
|
fail("confluence_target_ambiguous")
|
|
if not valid:
|
|
if read_only:
|
|
fail("confluence_target_missing")
|
|
initial = managed_section(
|
|
[], time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
|
)
|
|
created = client.request(
|
|
"POST",
|
|
"/wiki/api/v2/pages",
|
|
{
|
|
"spaceId": space_id,
|
|
"status": "current",
|
|
"title": CONFLUENCE_TITLE,
|
|
"parentId": CONFLUENCE_FOLDER_ID,
|
|
"body": {"representation": "storage", "value": initial},
|
|
},
|
|
)
|
|
page_id = str(created.get("id") or "")
|
|
if not page_id:
|
|
fail("confluence_create_invalid")
|
|
valid = [
|
|
client.request("GET", f"/wiki/api/v2/pages/{page_id}?body-format=storage")
|
|
]
|
|
if not page_in_folder(client, valid[0]):
|
|
fail("confluence_parent_invalid")
|
|
page = valid[0]
|
|
if page.get("title") != CONFLUENCE_TITLE or page.get("status") != "current":
|
|
fail("confluence_target_invalid")
|
|
confluence_storage(page)
|
|
page_version(page)
|
|
if not read_only:
|
|
canonical = (
|
|
f"{ATLASSIAN_BASE_URL}/wiki/spaces/Lab2/pages/{page['id']}\n".encode()
|
|
)
|
|
target_candidate = write_candidate(target_cache, canonical)
|
|
os.replace(target_candidate, target_cache)
|
|
os.chmod(target_cache, 0o600)
|
|
require_secure_secret(root, target_cache)
|
|
return page
|
|
|
|
|
|
def table_rows_from_store(
|
|
store_entries: dict[str, str], remote_mappings: list[dict[str, Any]]
|
|
) -> list[dict[str, str]]:
|
|
rows: list[dict[str, str]] = []
|
|
for principal_ref, raw_token in store_entries.items():
|
|
matches = [
|
|
item
|
|
for item in remote_mappings
|
|
if str(item.get("principal_ref")) == principal_ref
|
|
]
|
|
if len(matches) != 1:
|
|
fail("store_edge_mapping_mismatch")
|
|
item = matches[0]
|
|
if sha256_text(raw_token) != str(item.get("token_hash_sha256", "")).lower():
|
|
fail("store_edge_hash_mismatch")
|
|
rows.append(
|
|
{
|
|
"principal_ref": principal_ref,
|
|
"principal_alias": str(item.get("principal_alias", "")),
|
|
"token_ref": str(item.get("token_ref", "")),
|
|
}
|
|
)
|
|
if len({row["principal_ref"] for row in rows}) != len(rows):
|
|
fail("store_edge_mapping_duplicate")
|
|
return rows
|
|
|
|
|
|
def sync_confluence(
|
|
client: ConfluenceClient,
|
|
page: dict[str, Any],
|
|
mapping_rows: list[dict[str, str]],
|
|
) -> tuple[int, int]:
|
|
page_id = str(page.get("id") or "")
|
|
latest = client.request(
|
|
"GET", f"/wiki/api/v2/pages/{urllib.parse.quote(page_id)}?body-format=storage"
|
|
)
|
|
if not page_in_folder(client, latest):
|
|
fail("confluence_parent_invalid")
|
|
old_version = page_version(latest)
|
|
old_parent = str(latest.get("parentId") or "")
|
|
old_title = str(latest.get("title") or "")
|
|
old_status = str(latest.get("status") or "")
|
|
if old_title != CONFLUENCE_TITLE or old_status != "current":
|
|
fail("confluence_target_changed")
|
|
old_body = confluence_storage(latest)
|
|
records = parse_managed_table(old_body)
|
|
ensure_create_preserves_rows(records, mapping_rows)
|
|
try:
|
|
verify_managed_table(old_body, mapping_rows)
|
|
return old_version, len(mapping_rows)
|
|
except WorkflowFailure as error:
|
|
if error.code != "confluence_table_mismatch":
|
|
raise
|
|
synchronized_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
|
replacement = managed_section(mapping_rows, synchronized_at)
|
|
new_body = replace_managed_section(old_body, replacement)
|
|
client.request(
|
|
"PUT",
|
|
f"/wiki/api/v2/pages/{urllib.parse.quote(page_id)}",
|
|
{
|
|
"id": page_id,
|
|
"status": old_status,
|
|
"title": old_title,
|
|
"body": {"representation": "storage", "value": new_body},
|
|
"version": {"number": old_version + 1},
|
|
},
|
|
)
|
|
verified = client.request(
|
|
"GET", f"/wiki/api/v2/pages/{urllib.parse.quote(page_id)}?body-format=storage"
|
|
)
|
|
if (
|
|
str(verified.get("id")) != page_id
|
|
or str(verified.get("title")) != old_title
|
|
or str(verified.get("status")) != old_status
|
|
or str(verified.get("parentId") or "") != old_parent
|
|
or page_version(verified) != old_version + 1
|
|
):
|
|
fail("confluence_reread_mismatch")
|
|
verify_managed_table(confluence_storage(verified), mapping_rows)
|
|
return old_version + 1, len(mapping_rows)
|
|
|
|
|
|
def api_json(
|
|
root: Path,
|
|
profile: dict[str, Any],
|
|
path: str,
|
|
raw_token: str,
|
|
payload: dict[str, Any] | None = None,
|
|
timeout: int = 30,
|
|
) -> dict[str, Any]:
|
|
method = "POST" if payload is not None else "GET"
|
|
try:
|
|
status_code, data = curl_request(
|
|
root,
|
|
method,
|
|
str(profile["openai_base_url"]) + path,
|
|
[f"Authorization: Bearer {raw_token}", "Content-Type: application/json"],
|
|
payload,
|
|
timeout=timeout,
|
|
)
|
|
except CurlTransportFailure:
|
|
fail("openai_smoke_network_failed")
|
|
if status_code < 200 or status_code >= 300:
|
|
fail("openai_smoke_http_failed")
|
|
try:
|
|
value = json.loads(data)
|
|
except json.JSONDecodeError:
|
|
fail("openai_smoke_response_invalid")
|
|
if not isinstance(value, dict):
|
|
fail("openai_smoke_response_invalid")
|
|
return value
|
|
|
|
|
|
def api_smoke(root: Path, profile: dict[str, Any], raw_token: str) -> None:
|
|
if profile["openai_smoke_transport"] == "ssh-loopback":
|
|
remote_call(
|
|
root,
|
|
profile,
|
|
{"action": "api-smoke", "raw_token": raw_token},
|
|
timeout=135,
|
|
)
|
|
return
|
|
if profile["openai_smoke_transport"] != "public-https":
|
|
fail("openai_smoke_transport_invalid")
|
|
models = api_json(root, profile, "/models", raw_token, timeout=15)
|
|
if not isinstance(models.get("data"), list) or not models["data"]:
|
|
fail("openai_models_invalid")
|
|
response = api_json(
|
|
root,
|
|
profile,
|
|
"/chat/completions",
|
|
raw_token,
|
|
{
|
|
"model": profile["smoke_model"],
|
|
"messages": [{"role": "user", "content": "Reply with the single word OK."}],
|
|
"max_tokens": 2048,
|
|
"temperature": 0,
|
|
},
|
|
timeout=120,
|
|
)
|
|
choices = response.get("choices")
|
|
if not isinstance(choices, list) or not choices or not isinstance(choices[0], dict):
|
|
fail("openai_chat_invalid")
|
|
message = choices[0].get("message")
|
|
content = message.get("content") if isinstance(message, dict) else None
|
|
if (
|
|
not isinstance(content, str)
|
|
or not content.strip()
|
|
or not choices[0].get("finish_reason")
|
|
):
|
|
fail("openai_chat_invalid")
|
|
|
|
|
|
def metrics_observed(
|
|
root: Path, profile: dict[str, Any], token_ref: str
|
|
) -> bool | None:
|
|
if profile["metrics_transport"] == "ssh-loopback":
|
|
try:
|
|
result = remote_call(
|
|
root,
|
|
profile,
|
|
{"action": "metrics", "token_ref": token_ref},
|
|
timeout=15,
|
|
)
|
|
except WorkflowFailure:
|
|
return None
|
|
observed = result.get("observed")
|
|
return observed if isinstance(observed, bool) else None
|
|
if profile["metrics_transport"] != "public-http":
|
|
return None
|
|
try:
|
|
with urllib.request.urlopen(str(profile["metrics_url"]), timeout=8) as response:
|
|
metrics = response.read().decode("utf-8", "replace")
|
|
except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError):
|
|
return None
|
|
return re.search(r'token_ref="' + re.escape(token_ref) + r'"', metrics) is not None
|
|
|
|
|
|
def tracked_leak_check(root: Path, raw_token: str) -> None:
|
|
files = run_quiet(["git", "ls-files", "-z"], cwd=root)
|
|
if files.returncode != 0:
|
|
fail("tracked_file_list_failed")
|
|
needle = raw_token.encode("utf-8")
|
|
for item in files.stdout.split(b"\0"):
|
|
if not item:
|
|
continue
|
|
path = root / os.fsdecode(item)
|
|
try:
|
|
if path.is_file() and needle in path.read_bytes():
|
|
fail("raw_token_leak_detected")
|
|
except OSError:
|
|
fail("tracked_leak_check_failed")
|
|
|
|
|
|
def copy_once_to_clipboard(raw_token: str, root: Path) -> None:
|
|
copied = run_quiet(
|
|
["/usr/bin/pbcopy"], cwd=root, input_bytes=raw_token.encode("utf-8")
|
|
)
|
|
if copied.returncode != 0:
|
|
fail("clipboard_copy_failed")
|
|
pasted = run_quiet(["/usr/bin/pbpaste"], cwd=root)
|
|
if (
|
|
pasted.returncode != 0
|
|
or hashlib.sha256(pasted.stdout).digest()
|
|
!= hashlib.sha256(raw_token.encode()).digest()
|
|
):
|
|
fail("clipboard_verify_failed")
|
|
|
|
|
|
def load_confluence(
|
|
root: Path, profile: dict[str, Any]
|
|
) -> tuple[ConfluenceClient, Path]:
|
|
token_path = root / "token/.lgu-atlassian-token"
|
|
username_path = root / str(profile["confluence_user_store"])
|
|
target_cache = root / "token/.dev-corp-iop-confluence-target"
|
|
require_secure_secret(root, token_path)
|
|
require_secure_secret(root, username_path)
|
|
token = token_path.read_text(encoding="utf-8").strip()
|
|
username = username_path.read_text(encoding="utf-8").strip()
|
|
if not token or "\n" in token or "\r" in token:
|
|
fail("atlassian_token_invalid")
|
|
if (
|
|
not username
|
|
or "\n" in username
|
|
or "\r" in username
|
|
or not re.fullmatch(r"[^@\s]+@[^@\s]+", username)
|
|
):
|
|
fail("atlassian_user_invalid")
|
|
if target_cache.exists():
|
|
require_secure_secret(root, target_cache)
|
|
elif not git_is_ignored(root, target_cache):
|
|
fail("target_cache_ignore_invalid")
|
|
return ConfluenceClient(root, username, token), target_cache
|
|
|
|
|
|
def preflight(request: dict[str, str]) -> dict[str, Any]:
|
|
root = repo_root()
|
|
profile = load_profile(root, request["env"])
|
|
store_path = root / str(profile["secret_store"])
|
|
require_secure_secret(root, store_path, must_exist=False)
|
|
_store_bytes, store_entries = read_store(store_path)
|
|
remote = remote_call(
|
|
root,
|
|
profile,
|
|
{"action": "inspect", "refresh_probe": True},
|
|
timeout=20,
|
|
)
|
|
mappings = remote.get("mappings")
|
|
if (
|
|
not isinstance(mappings, list)
|
|
or not remote.get("candidate_shape")
|
|
or remote.get("refresh_status") != "applied"
|
|
or not remote.get("healthy")
|
|
or not remote.get("listener")
|
|
):
|
|
fail("edge_preflight_invalid")
|
|
remote_mappings = [item for item in mappings if isinstance(item, dict)]
|
|
confluence_status = "not-applicable"
|
|
if profile["confluence_enabled"]:
|
|
client, target_cache = load_confluence(root, profile)
|
|
page = preflight_confluence(root, client, target_cache, read_only=True)
|
|
current_rows = table_rows_from_store(store_entries, remote_mappings)
|
|
ensure_create_preserves_rows(
|
|
parse_managed_table(confluence_storage(page)), current_rows
|
|
)
|
|
confluence_status = "ready"
|
|
request_state = (
|
|
"existing"
|
|
if select_existing(request, store_entries, remote_mappings)
|
|
else "new"
|
|
)
|
|
return {
|
|
"result": "ready",
|
|
"mode": "preflight",
|
|
"environment": request["env"],
|
|
"edge": "healthy",
|
|
"confluence": confluence_status,
|
|
"stored_user_count": len(store_entries),
|
|
"request_state": request_state,
|
|
"raw_token_generated": False,
|
|
}
|
|
|
|
|
|
def select_existing(
|
|
request: dict[str, str],
|
|
store_entries: dict[str, str],
|
|
remote_mappings: list[dict[str, Any]],
|
|
) -> tuple[dict[str, Any], str] | None:
|
|
principal_matches = [
|
|
item
|
|
for item in remote_mappings
|
|
if str(item.get("principal_ref")) == request["principal_ref"]
|
|
]
|
|
ref_matches = [
|
|
item
|
|
for item in remote_mappings
|
|
if str(item.get("token_ref")) == request["token_ref"]
|
|
]
|
|
local_token = store_entries.get(request["principal_ref"])
|
|
if len(principal_matches) > 1 or len(ref_matches) > 1:
|
|
fail("existing_mapping_ambiguous")
|
|
if ref_matches and (
|
|
not principal_matches or ref_matches[0] is not principal_matches[0]
|
|
):
|
|
fail("token_ref_conflict")
|
|
if principal_matches:
|
|
if local_token is None:
|
|
fail("existing_mapping_missing_local_token")
|
|
mapping = principal_matches[0]
|
|
if (
|
|
sha256_text(local_token)
|
|
!= str(mapping.get("token_hash_sha256", "")).lower()
|
|
):
|
|
fail("existing_mapping_hash_mismatch")
|
|
return mapping, local_token
|
|
if local_token is not None:
|
|
fail("local_token_missing_edge_mapping")
|
|
return None
|
|
|
|
|
|
def execute(request: dict[str, str]) -> dict[str, Any]:
|
|
started = time.monotonic()
|
|
root = repo_root()
|
|
profile = load_profile(root, request["env"])
|
|
store_path = root / str(profile["secret_store"])
|
|
require_secure_secret(root, store_path, must_exist=False)
|
|
old_store_bytes, store_entries = read_store(store_path)
|
|
client: ConfluenceClient | None = None
|
|
target_cache: Path | None = None
|
|
page: dict[str, Any] | None = None
|
|
|
|
print("[1/6] preflight", file=sys.stderr, flush=True)
|
|
if profile["confluence_enabled"]:
|
|
client, target_cache = load_confluence(root, profile)
|
|
page = preflight_confluence(root, client, target_cache, read_only=False)
|
|
remote = remote_call(
|
|
root,
|
|
profile,
|
|
{"action": "inspect", "refresh_probe": True},
|
|
timeout=20,
|
|
)
|
|
mappings_value = remote.get("mappings")
|
|
if (
|
|
not isinstance(mappings_value, list)
|
|
or not remote.get("candidate_shape")
|
|
or remote.get("refresh_status") != "applied"
|
|
or not remote.get("healthy")
|
|
or not remote.get("listener")
|
|
):
|
|
fail("edge_preflight_invalid")
|
|
remote_mappings = [item for item in mappings_value if isinstance(item, dict)]
|
|
if page is not None:
|
|
current_rows = table_rows_from_store(store_entries, remote_mappings)
|
|
ensure_create_preserves_rows(
|
|
parse_managed_table(confluence_storage(page)), current_rows
|
|
)
|
|
existing = select_existing(request, store_entries, remote_mappings)
|
|
|
|
edge_state = "existing-verified"
|
|
backup_path: str | None = None
|
|
raw_token: str
|
|
if existing:
|
|
mapping, raw_token = existing
|
|
request["principal_alias"] = str(
|
|
mapping.get("principal_alias") or request["principal_alias"]
|
|
)
|
|
request["token_ref"] = str(mapping.get("token_ref") or request["token_ref"])
|
|
print("[2/6] existing mapping reused", file=sys.stderr, flush=True)
|
|
else:
|
|
print("[2/6] token generated in memory", file=sys.stderr, flush=True)
|
|
raw_token = "iop_" + secrets.token_urlsafe(36)
|
|
token_hash = sha256_text(raw_token)
|
|
new_content = old_store_bytes
|
|
if new_content and not new_content.endswith(b"\n"):
|
|
new_content += b"\n"
|
|
new_content += f"{request['principal_ref']}: {raw_token}\n".encode()
|
|
local_candidate = write_candidate(store_path, new_content)
|
|
try:
|
|
print("[3/6] Edge transaction", file=sys.stderr, flush=True)
|
|
activated = remote_call(
|
|
root,
|
|
profile,
|
|
{
|
|
"action": "apply",
|
|
"entry": {
|
|
"token_ref": request["token_ref"],
|
|
"token_hash_sha256": token_hash,
|
|
"principal_ref": request["principal_ref"],
|
|
"principal_alias": request["principal_alias"],
|
|
},
|
|
},
|
|
timeout=90,
|
|
)
|
|
backup_path = str(activated.get("backup_path") or "")
|
|
if not backup_path:
|
|
fail("edge_backup_reference_missing")
|
|
os.replace(local_candidate, store_path)
|
|
os.chmod(store_path, 0o600)
|
|
require_secure_secret(root, store_path)
|
|
edge_state = "activated"
|
|
except BaseException:
|
|
if backup_path:
|
|
rollback = remote_call(
|
|
root,
|
|
profile,
|
|
{"action": "rollback", "backup_path": backup_path},
|
|
timeout=70,
|
|
)
|
|
if rollback.get("status") != "rolled_back":
|
|
fail("edge_rollback_unconfirmed")
|
|
atomic_restore(store_path, old_store_bytes if old_store_bytes else None)
|
|
raise
|
|
finally:
|
|
local_candidate.unlink(missing_ok=True)
|
|
|
|
print("[4/6] environment API smoke", file=sys.stderr, flush=True)
|
|
try:
|
|
api_smoke(root, profile, raw_token)
|
|
except WorkflowFailure:
|
|
if backup_path:
|
|
rollback = remote_call(
|
|
root,
|
|
profile,
|
|
{"action": "rollback", "backup_path": backup_path},
|
|
timeout=70,
|
|
)
|
|
if rollback.get("status") != "rolled_back":
|
|
fail("edge_rollback_unconfirmed")
|
|
atomic_restore(store_path, old_store_bytes if old_store_bytes else None)
|
|
raise
|
|
|
|
version: int | None = None
|
|
confluence_status = "not-applicable"
|
|
_current_bytes, current_store = read_store(store_path)
|
|
row_count = len(current_store)
|
|
if client is not None and page is not None:
|
|
print("[5/6] Confluence metadata sync", file=sys.stderr, flush=True)
|
|
current_remote = remote_call(root, profile, {"action": "inspect"}, timeout=20)
|
|
current_values = current_remote.get("mappings")
|
|
if not isinstance(current_values, list):
|
|
fail("edge_mapping_reread_invalid")
|
|
rows = table_rows_from_store(
|
|
current_store, [item for item in current_values if isinstance(item, dict)]
|
|
)
|
|
version, row_count = sync_confluence(client, page, rows)
|
|
confluence_status = "updated"
|
|
else:
|
|
print("[5/6] Confluence not applicable", file=sys.stderr, flush=True)
|
|
|
|
print("[6/6] leak check and one-time delivery", file=sys.stderr, flush=True)
|
|
tracked_leak_check(root, raw_token)
|
|
require_secure_secret(root, store_path)
|
|
if target_cache is not None:
|
|
require_secure_secret(root, target_cache)
|
|
copy_once_to_clipboard(raw_token, root)
|
|
observed = metrics_observed(root, profile, request["token_ref"])
|
|
return {
|
|
"result": "completed",
|
|
"operation": "create",
|
|
"environment": request["env"],
|
|
"principal_alias": request["principal_alias"],
|
|
"token_ref": request["token_ref"],
|
|
"edge_mapping": edge_state,
|
|
"api_smoke": "passed",
|
|
"confluence_metadata_sync": confluence_status,
|
|
"confluence_version": version,
|
|
"metadata_row_count": row_count,
|
|
"metrics_observed": observed,
|
|
"raw_token_delivered_once": True,
|
|
"delivery_channel": "local_clipboard",
|
|
"leak_check": "passed",
|
|
"elapsed_seconds": round(time.monotonic() - started, 1),
|
|
}
|
|
|
|
|
|
def selftest() -> dict[str, Any]:
|
|
sample = normalize_alias("A.Name+test@example.invalid", None)
|
|
if sample != "a-name-test":
|
|
fail("selftest_alias_failed")
|
|
if normalize_alias("a@example.invalid", None) != "a":
|
|
fail("selftest_short_alias_failed")
|
|
try:
|
|
parse_request('{"env":"dev","principal_ref":null}')
|
|
except WorkflowFailure as error:
|
|
if error.code != "principal_ref_invalid":
|
|
raise
|
|
else:
|
|
fail("selftest_input_type_failed")
|
|
try:
|
|
parse_request('{"principal_ref":"sample@example.invalid"}')
|
|
except WorkflowFailure as error:
|
|
if error.code != "environment_not_supported":
|
|
raise
|
|
else:
|
|
fail("selftest_environment_required_failed")
|
|
root = repo_root()
|
|
dev_request = parse_request(
|
|
'{"env":"dev","principal_ref":"sample@example.invalid"}'
|
|
)
|
|
corp_request = parse_request(
|
|
'{"env":"dev-corp","principal_ref":"sample@example.invalid"}'
|
|
)
|
|
if dev_request["token_ref"] != "iop-dev-sample":
|
|
fail("selftest_dev_token_ref_failed")
|
|
if corp_request["token_ref"] != "iop-dev-corp-sample":
|
|
fail("selftest_dev_corp_token_ref_failed")
|
|
for environment in SUPPORTED_ENVIRONMENTS:
|
|
profile = load_profile(root, environment)
|
|
if profile["environment"] != environment:
|
|
fail("selftest_profile_failed")
|
|
rows = [
|
|
{
|
|
"principal_ref": "sample@example.invalid",
|
|
"principal_alias": "sample",
|
|
"token_ref": "iop-dev-corp-sample",
|
|
}
|
|
]
|
|
original = "<h1>Intro</h1><p>Keep</p><h2>IOP 사용자 토큰 발급 현황</h2><p>old</p><h2>Next</h2><p>Keep too</p>"
|
|
replaced = replace_managed_section(
|
|
original, managed_section(rows, "2026-01-01T00:00:00Z")
|
|
)
|
|
verify_managed_table(replaced, rows)
|
|
if "<p>Keep</p>" not in replaced or "<p>Keep too</p>" not in replaced:
|
|
fail("selftest_section_failed")
|
|
records = parse_managed_table(replaced)
|
|
try:
|
|
ensure_create_preserves_rows(records, [])
|
|
except WorkflowFailure as error:
|
|
if error.code != "confluence_create_would_remove_user":
|
|
raise
|
|
else:
|
|
fail("selftest_append_only_failed")
|
|
helper = (
|
|
root
|
|
/ "agent-ops/skills/project/openai-usage-token-issue/scripts/remote_edge_transaction.rb"
|
|
)
|
|
for environment in SUPPORTED_ENVIRONMENTS:
|
|
ruby = run_quiet(
|
|
["ruby", str(helper)],
|
|
cwd=root,
|
|
input_bytes=json.dumps(
|
|
{"action": "selftest", "environment": environment},
|
|
separators=(",", ":"),
|
|
).encode(),
|
|
timeout=10,
|
|
)
|
|
parsed = None
|
|
for line in reversed(ruby.stdout.decode("utf-8", "replace").splitlines()):
|
|
try:
|
|
value = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if isinstance(value, dict):
|
|
parsed = value
|
|
break
|
|
if ruby.returncode != 0 or not parsed or parsed.get("status") != "ok":
|
|
fail("selftest_remote_helper_failed")
|
|
return {"result": "passed", "mode": "self-test", "network_used": False}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description="Issue one dev or dev-corp IOP user token from stdin JSON."
|
|
)
|
|
mode = parser.add_mutually_exclusive_group()
|
|
mode.add_argument(
|
|
"--preflight", action="store_true", help="Run read-only live readiness checks."
|
|
)
|
|
mode.add_argument(
|
|
"--self-test", action="store_true", help="Run local deterministic tests."
|
|
)
|
|
arguments = parser.parse_args()
|
|
try:
|
|
if arguments.self_test:
|
|
result = selftest()
|
|
else:
|
|
request = parse_request(sys.stdin.read())
|
|
root = repo_root()
|
|
with local_transaction_lock(root, shared=arguments.preflight):
|
|
result = preflight(request) if arguments.preflight else execute(request)
|
|
print(json.dumps(result, ensure_ascii=False, separators=(",", ":")))
|
|
return 0
|
|
except WorkflowFailure as error:
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"result": "blocked",
|
|
"code": error.code,
|
|
"raw_token_reported": False,
|
|
"retry_same_command": True,
|
|
},
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
)
|
|
)
|
|
return 2
|
|
except KeyboardInterrupt:
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"result": "blocked",
|
|
"code": "interrupted",
|
|
"raw_token_reported": False,
|
|
"retry_same_command": True,
|
|
},
|
|
separators=(",", ":"),
|
|
)
|
|
)
|
|
return 130
|
|
except Exception: # noqa: BLE001 - never render a traceback that could retain secret request state.
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"result": "blocked",
|
|
"code": "unexpected_local_failure",
|
|
"raw_token_reported": False,
|
|
"retry_same_command": True,
|
|
},
|
|
separators=(",", ":"),
|
|
)
|
|
)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|