"""refresh_bot_token.py — GitHub App installation token refresh (50-min cycle). 회장 박제 명세 feedback_github_app_key_location_260507.md 준수. stdlib + PyJWT + cryptography 만 사용 (requests/httpx 금지). """ from __future__ import annotations import argparse import hashlib import json import os import tempfile import time import urllib.error import urllib.request from pathlib import Path from typing import Optional import jwt # PyJWT 2.x # --------------------------------------------------------------------------- # 공개 상수 (Morrigan 회귀 테스트 import 대상) # --------------------------------------------------------------------------- WORKSPACE = Path("/home/jay/workspace") DEFAULT_PEM_BACKUP = WORKSPACE / ".secrets/jeon-jonghyuk-taskctl-bot.2026-05-05.private-key.pem" DEFAULT_AUDIT_PATH = WORKSPACE / "memory/orchestration-audit/bot-token-refresh.jsonl" DEFAULT_ENV_KEYS_PATH = WORKSPACE / ".env.keys" GITHUB_API_BASE = "https://api.github.com" # --------------------------------------------------------------------------- # .env.keys 파서 (export prefix strip 지원) # --------------------------------------------------------------------------- def _parse_env_keys(env_path: Path) -> dict[str, str]: """라인별 NAME=VALUE 파싱. 'export ' prefix strip.""" result: dict[str, str] = {} if not env_path.exists(): return result try: for raw_line in env_path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#"): continue # export prefix strip if line.startswith("export "): line = line[len("export "):].strip() if "=" not in line: continue k, v = line.split("=", 1) k = k.strip() v = v.strip().strip('"').strip("'") if k: result[k] = v except Exception: pass return result def _getenv(key: str, env_keys_path: Path) -> Optional[str]: """os.environ 우선, 없으면 .env.keys 파싱.""" val = os.environ.get(key) if val: return val parsed = _parse_env_keys(env_keys_path) return parsed.get(key) # --------------------------------------------------------------------------- # 공개 인터페이스 # --------------------------------------------------------------------------- def resolve_pem_path( env_path: Optional[str], fallback_path: Path = DEFAULT_PEM_BACKUP, ) -> Path: """env BOT_GITHUB_PRIVATE_KEY_PATH 우선 → 부재 시 fallback. 둘 다 부재(파일 없음)면 FileNotFoundError.""" # 메인 경로: /home/jay/.secrets/... (env 또는 fallback에서 얻음) main_pem = Path("/home/jay/.secrets/jeon-jonghyuk-taskctl-bot.2026-05-05.private-key.pem") candidates: list[Path] = [] if env_path: candidates.append(Path(env_path)) candidates.append(main_pem) candidates.append(fallback_path) for p in candidates: if p.exists(): return p raise FileNotFoundError( f"PEM key not found. Tried: {[str(c) for c in candidates]}" ) def generate_jwt( app_id: str, pem_bytes: bytes, *, now: Optional[int] = None, ) -> str: """RS256, iat=now-60 (시계 보정), exp=now+540 (9분). PyJWT 사용.""" if now is None: now = int(time.time()) payload = { "iat": now - 60, "exp": now + 540, "iss": app_id, } return jwt.encode(payload, pem_bytes, algorithm="RS256") def request_installation_token( jwt_token: str, installation_id: str, *, timeout: float = 10.0, ) -> dict: """POST /app/installations/{id}/access_tokens. 200 → {"token": ..., "expires_at": ...}. 비-200 → raise RuntimeError(status, body). """ url = f"{GITHUB_API_BASE}/app/installations/{installation_id}/access_tokens" req = urllib.request.Request( url, method="POST", data=b"", headers={ "Authorization": f"Bearer {jwt_token}", "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28", "Content-Length": "0", }, ) try: with urllib.request.urlopen(req, timeout=timeout) as resp: body = resp.read().decode("utf-8") return json.loads(body) except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") raise RuntimeError(exc.code, body) from exc def update_env_keys(env_path: Path, new_token: str) -> None: """라인이 'BOT_GITHUB_TOKEN='로 시작하면 in-place 교체. 미존재 시 EOF append. atomic (tmpfile + os.replace). chmod 0o600 보존. export prefix 추가 X. """ target_prefix = "BOT_GITHUB_TOKEN=" # 파일 존재 여부 / 기존 내용 읽기 if env_path.exists(): original_text = env_path.read_text(encoding="utf-8") # 기존 권한 보존 original_mode = env_path.stat().st_mode & 0o777 else: original_text = "" original_mode = 0o600 lines = original_text.splitlines(keepends=True) replaced = False new_lines: list[str] = [] for line in lines: if line.startswith(target_prefix): new_lines.append(f"{target_prefix}{new_token}\n") replaced = True else: new_lines.append(line) if not replaced: # EOF append if new_lines and not new_lines[-1].endswith("\n"): new_lines.append("\n") new_lines.append(f"{target_prefix}{new_token}\n") new_text = "".join(new_lines) # atomic write: tmpfile → os.replace. # fd ownership: os.fdopen(...) takes ownership and closes via with-block. # 예외 발생 시에도 fd 누수가 없도록 try/finally + close-once 가드. env_path.parent.mkdir(parents=True, exist_ok=True) fd, tmp_path = tempfile.mkstemp( dir=env_path.parent, prefix=".env.keys.tmp." ) fd_closed = False try: with os.fdopen(fd, "w", encoding="utf-8") as f: fd_closed = True f.write(new_text) os.chmod(tmp_path, original_mode) os.replace(tmp_path, env_path) except Exception: if not fd_closed: try: os.close(fd) except OSError: pass try: os.unlink(tmp_path) except OSError: pass raise def append_audit( audit_path: Path, *, status: str, sha256_prefix: Optional[str], expires_at: Optional[str], error: Optional[str] = None, ) -> None: """jsonl append-only. status ∈ {refreshed, refresh_rejected, pem_missing, api_error, dry_run}. 토큰 원문 절대 기록 X. sha256_prefix는 hashlib.sha256(token.encode()).hexdigest()[:16]. """ audit_path.parent.mkdir(parents=True, exist_ok=True) record: dict = { "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "status": status, "sha256_prefix": sha256_prefix, "expires_at": expires_at, } if error is not None: record["error"] = error line = json.dumps(record, ensure_ascii=False) + "\n" with audit_path.open("a", encoding="utf-8") as f: f.write(line) # --------------------------------------------------------------------------- # CLI entry # --------------------------------------------------------------------------- def main(argv: Optional[list[str]] = None) -> int: """CLI entry. exit 0=성공, 1=실패. --dry-run flag 지원.""" parser = argparse.ArgumentParser( description="GitHub App installation token refresh" ) parser.add_argument( "--dry-run", action="store_true", help="토큰 발급만 수행, .env.keys 미수정", ) parser.add_argument( "--env-keys", default=str(DEFAULT_ENV_KEYS_PATH), help=f".env.keys 경로 (기본값: {DEFAULT_ENV_KEYS_PATH})", ) parser.add_argument( "--audit-path", default=str(DEFAULT_AUDIT_PATH), help=f"audit jsonl 경로 (기본값: {DEFAULT_AUDIT_PATH})", ) args = parser.parse_args(argv) env_keys_path = Path(args.env_keys) audit_path = Path(args.audit_path) dry_run: bool = args.dry_run # --- 1. 환경 변수 로드 --- app_id = _getenv("BOT_GITHUB_APP_ID", env_keys_path) installation_id = _getenv("BOT_GITHUB_INSTALLATION_ID", env_keys_path) pem_env_path = _getenv("BOT_GITHUB_PRIVATE_KEY_PATH", env_keys_path) if not app_id: print("[ERROR] BOT_GITHUB_APP_ID 미설정", flush=True) append_audit( audit_path, status="api_error", sha256_prefix=None, expires_at=None, error="BOT_GITHUB_APP_ID missing", ) return 1 if not installation_id: print("[ERROR] BOT_GITHUB_INSTALLATION_ID 미설정", flush=True) append_audit( audit_path, status="api_error", sha256_prefix=None, expires_at=None, error="BOT_GITHUB_INSTALLATION_ID missing", ) return 1 # --- 2. PEM 로드 --- try: pem_path = resolve_pem_path(pem_env_path) pem_bytes = pem_path.read_bytes() except FileNotFoundError as exc: print(f"[ERROR] PEM 파일 없음: {exc}", flush=True) append_audit( audit_path, status="pem_missing", sha256_prefix=None, expires_at=None, error=str(exc), ) return 1 # --- 3. JWT 생성 --- try: jwt_token = generate_jwt(app_id, pem_bytes) except Exception as exc: print(f"[ERROR] JWT 생성 실패: {exc}", flush=True) append_audit( audit_path, status="api_error", sha256_prefix=None, expires_at=None, error=f"jwt_error: {exc}", ) return 1 # --- 4. Installation token 요청 --- try: result = request_installation_token(jwt_token, installation_id) except RuntimeError as exc: status_code, body = exc.args print(f"[ERROR] GitHub API 오류 HTTP {status_code}: {body[:200]}", flush=True) audit_status = "refresh_rejected" if 400 <= int(status_code) < 500 else "api_error" append_audit( audit_path, status=audit_status, sha256_prefix=None, expires_at=None, error=f"http_{status_code}: {body[:200]}", ) return 1 except Exception as exc: print(f"[ERROR] API 호출 실패: {exc}", flush=True) append_audit( audit_path, status="api_error", sha256_prefix=None, expires_at=None, error=str(exc), ) return 1 # --- 5. 토큰 추출 --- new_token = result.get("token", "") expires_at = result.get("expires_at", "") if not new_token: print("[ERROR] 응답에 token 필드 없음", flush=True) append_audit( audit_path, status="api_error", sha256_prefix=None, expires_at=None, error="token field missing in response", ) return 1 sha256_prefix = hashlib.sha256(new_token.encode()).hexdigest()[:16] token_prefix = new_token[:12] # prefix 12자 (원문 출력 금지) # --- 6. dry-run 분기 --- if dry_run: append_audit( audit_path, status="dry_run", sha256_prefix=sha256_prefix, expires_at=expires_at, ) output = { "status": "dry_run", "token_prefix": token_prefix, "expires_at": expires_at, "sha256_prefix": sha256_prefix, "dry_run": True, } print(json.dumps(output, ensure_ascii=False), flush=True) return 0 # --- 7. .env.keys 갱신 --- try: update_env_keys(env_keys_path, new_token) except Exception as exc: print(f"[ERROR] .env.keys 갱신 실패: {exc}", flush=True) append_audit( audit_path, status="api_error", sha256_prefix=sha256_prefix, expires_at=expires_at, error=f"env_keys_write_error: {exc}", ) return 1 # --- 8. audit 기록 + stdout 출력 --- append_audit( audit_path, status="refreshed", sha256_prefix=sha256_prefix, expires_at=expires_at, ) output = { "status": "refreshed", "token_prefix": token_prefix, "expires_at": expires_at, "sha256_prefix": sha256_prefix, "dry_run": False, } print(json.dumps(output, ensure_ascii=False), flush=True) return 0 if __name__ == "__main__": raise SystemExit(main())