chore(harness): Claude Code 하네스 구성 — 안전 가드·규칙 검사·배포 스킬
- .claude/settings.json (공유) 신설, hooks 를 settings.local.json 에서 이동 - bash_guard (PreToolUse): 로컬 스택 기동·운영 DB alembic·_test 아닌 pytest·main 강제 push·위험 rm 차단, main push/deploy.sh 는 장중 경고와 함께 확인 요청, backend 커밋 전 유닛 스위트 게이트 - rules_check (PostToolUse): py_compile + CLAUDE.md 규칙(json.dumps, except 후 logger.debug, except-pass, 실시간 계층 DB I/O) — diff 추가 라인만 - session_context (SessionStart): 브랜치·미커밋·장 상태·alembic head 주입 - stop_check (Stop): frontend 변경 시 tsc - auto-orchestrate: 서브에이전트 보고·시스템 알림·질문형 프롬프트 오발동 제거 - /deploy 스킬 + deploy.sh / prod_smoke.sh: 테스트→dev push→main 병합→재기동 감시→공개 API 스모크 - 팀 에이전트·orchestrate 스킬 추적 시작, settings.local.json 추적 해제(.gitignore) - CLAUDE.md §9 하네스 - 훅 테스트 69 passed Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
364
.claude/hooks/bash_guard.py
Executable file
364
.claude/hooks/bash_guard.py
Executable file
@@ -0,0 +1,364 @@
|
||||
#!/usr/bin/env python3
|
||||
"""PreToolUse(Bash) 가드 — 운영 안전장치를 사람이 아니라 훅이 강제한다.
|
||||
|
||||
deny : 로컬 스택 기동, 운영 DB 대상 alembic, _test 아닌 DB 로 pytest, --allow-network,
|
||||
main 강제 푸시, 레포 핵심 경로 rm -rf, 유닛 스위트 실패 상태의 backend 커밋
|
||||
ask : main 갱신 푸시 / deploy.sh 실행 (KST 시각·장중 경고 포함)
|
||||
파싱 실패는 조용히 통과(exit 0). 인용부호 안의 문자열은 명령으로 보지 않는다.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shlex
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from _lib import git, kst_now, market_state, project_dir, read_stdin_json # noqa: E402
|
||||
|
||||
WRAPPERS = {"sudo", "env", "time", "nohup", "command", "exec", "nice", "builtin"}
|
||||
SHELLS = {"bash", "sh", "zsh", "dash"}
|
||||
PYTEST_CMD_DEFAULT = (
|
||||
'cd backend && DATABASE_URL="postgresql+asyncpg://x:x@localhost/autostock_test" '
|
||||
"../.venv/bin/python -m pytest tests/ --ignore=tests/integration -q -x -p no:cacheprovider"
|
||||
)
|
||||
|
||||
|
||||
# ───────────────────────── 파싱 ─────────────────────────
|
||||
def _strip_heredocs(cmd: str) -> str:
|
||||
out, lines, delim = [], cmd.split("\n"), None
|
||||
for ln in lines:
|
||||
if delim is not None:
|
||||
if ln.strip() == delim:
|
||||
delim = None
|
||||
continue
|
||||
out.append(ln)
|
||||
m = re.search(r"<<-?\s*['\"]?(\w+)['\"]?", ln)
|
||||
if m:
|
||||
delim = m.group(1)
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
def split_segments(cmd: str) -> list[str]:
|
||||
"""따옴표를 존중하며 ; && || | & 줄바꿈 ( ) ` 로 분리."""
|
||||
segs, cur, q, i, n = [], [], None, 0, len(cmd)
|
||||
while i < n:
|
||||
c = cmd[i]
|
||||
if q:
|
||||
cur.append(c)
|
||||
if c == "\\" and q == '"' and i + 1 < n:
|
||||
cur.append(cmd[i + 1])
|
||||
i += 1
|
||||
elif c == q:
|
||||
q = None
|
||||
elif c == "\\" and i + 1 < n:
|
||||
cur.append(c + cmd[i + 1])
|
||||
i += 1
|
||||
elif c in "'\"":
|
||||
q = c
|
||||
cur.append(c)
|
||||
elif c in ";|&\n()`":
|
||||
segs.append("".join(cur))
|
||||
cur = []
|
||||
else:
|
||||
cur.append(c)
|
||||
i += 1
|
||||
segs.append("".join(cur))
|
||||
return [s.strip() for s in segs if s.strip()]
|
||||
|
||||
|
||||
def tokenize(seg: str) -> list[str]:
|
||||
try:
|
||||
return shlex.split(seg, posix=True)
|
||||
except ValueError:
|
||||
return seg.split()
|
||||
|
||||
|
||||
def split_assignments(argv: list[str]) -> tuple[dict, list[str]]:
|
||||
env, i = {}, 0
|
||||
while i < len(argv):
|
||||
t = argv[i]
|
||||
m = re.match(r"^([A-Za-z_][A-Za-z0-9_]*)=(.*)$", t)
|
||||
if m:
|
||||
env[m.group(1)] = m.group(2)
|
||||
i += 1
|
||||
elif t in WRAPPERS:
|
||||
i += 1
|
||||
elif t == "export" and i + 1 < len(argv):
|
||||
i += 1
|
||||
else:
|
||||
break
|
||||
return env, argv[i:]
|
||||
|
||||
|
||||
def base(tok: str) -> str:
|
||||
return os.path.basename(tok)
|
||||
|
||||
|
||||
# ───────────────────────── 판정 ─────────────────────────
|
||||
def deny(reason: str) -> dict:
|
||||
return {"decision": "deny", "reason": reason}
|
||||
|
||||
|
||||
def ask(reason: str) -> dict:
|
||||
return {"decision": "ask", "reason": reason}
|
||||
|
||||
|
||||
def is_pytest(argv: list[str]) -> bool:
|
||||
return any(base(t) in ("pytest", "py.test") for t in argv[:4]) or (
|
||||
"-m" in argv and "pytest" in argv
|
||||
)
|
||||
|
||||
|
||||
def is_prod_url(url: str) -> bool:
|
||||
low = url.lower()
|
||||
return any(k in low for k in ("nas", "15433", "autostock")) and not url.split("?")[0].endswith("_test")
|
||||
|
||||
|
||||
def check_local_stack(argv: list[str]) -> dict | None:
|
||||
b0 = base(argv[0]) if argv else ""
|
||||
names = [base(t) for t in argv]
|
||||
uvicorn_main = ("uvicorn" in names or b0 == "gunicorn") and any(
|
||||
t.startswith("app.main") or t.startswith("app:") for t in argv
|
||||
)
|
||||
py_dash_m = "-m" in argv and argv.index("-m") + 1 < len(argv) and argv[argv.index("-m") + 1].startswith(
|
||||
("app.main", "uvicorn", "gunicorn")
|
||||
)
|
||||
if uvicorn_main or py_dash_m:
|
||||
return deny(
|
||||
"로컬 백엔드 기동 금지: NAS 와 KIS 앱키를 공유하므로 토큰이 무효화된다"
|
||||
"(local-docker-removed-kis-token-conflict 메모리). 검증은 격리 pytest(_test DB)로만 하라."
|
||||
)
|
||||
if b0 == "fastapi" and any(t in ("run", "dev") for t in argv[1:3]):
|
||||
return deny("로컬 fastapi 기동 금지(KIS 토큰 충돌).")
|
||||
if b0 == "docker-compose" or (b0 == "docker" and "compose" in argv):
|
||||
start = argv.index("compose") if "compose" in argv else 0
|
||||
if any(t in ("up", "start", "restart") for t in argv[start + 1:]):
|
||||
return deny(
|
||||
"로컬 docker compose up/start/restart 금지: KIS 앱키 공유로 NAS 토큰이 무효화된다. "
|
||||
"배포는 /deploy 스킬(Jenkins), 검증은 격리 pytest."
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def check_alembic(argv: list[str], env: dict) -> dict | None:
|
||||
if not (any(base(t) == "alembic" for t in argv) or ("-m" in argv and "alembic" in argv)):
|
||||
return None
|
||||
if not any(t in ("upgrade", "downgrade") for t in argv):
|
||||
return None
|
||||
if base(argv[0]) == "ssh":
|
||||
return None
|
||||
url = env.get("DATABASE_URL", os.environ.get("DATABASE_URL", ""))
|
||||
if url and is_prod_url(url):
|
||||
return deny("운영 DB(nas/15433/autostock, _test 아님)를 대상으로 로컬 alembic upgrade/downgrade 금지. 마이그레이션은 배포 파이프라인(Jenkins)이 앱 기동 전에 실행한다.")
|
||||
if not url:
|
||||
return ask("DATABASE_URL 미지정 alembic upgrade/downgrade — .env 가 운영 DB 를 가리킬 수 있다. 대상 DB 가 *_test 인지 확인했는가?")
|
||||
return None
|
||||
|
||||
|
||||
def check_pytest(argv: list[str], env: dict) -> dict | None:
|
||||
if not is_pytest(argv):
|
||||
return None
|
||||
if "--allow-network" in argv:
|
||||
return deny("--allow-network 금지: KIS 호출 차단을 끄면 실주문이 나갈 수 있다(CLAUDE.md §4.1, 2026-09-27 실주문 2회 사고).")
|
||||
if any(".claude/hooks" in t for t in argv):
|
||||
return None # 하네스 훅 자체 테스트는 DB 무관
|
||||
url = env.get("DATABASE_URL", os.environ.get("DATABASE_URL", ""))
|
||||
if not url.split("?")[0].endswith("_test"):
|
||||
return deny(
|
||||
"pytest 는 DATABASE_URL 이 *_test 로 끝나야 한다(CLAUDE.md §4.1). 예: "
|
||||
'cd backend && DATABASE_URL="postgresql+asyncpg://x:x@localhost/autostock_test" '
|
||||
"../.venv/bin/python -m pytest tests/ --ignore=tests/integration"
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def check_git_push(argv: list[str], cwd: str) -> dict | None:
|
||||
if base(argv[0]) != "git" or "push" not in argv:
|
||||
return None
|
||||
args = argv[argv.index("push") + 1:]
|
||||
force = any(a in ("-f", "--force", "--force-with-lease") or a.startswith("--force-with-lease=") for a in args) \
|
||||
or any(re.match(r"^-[a-zA-Z]*f[a-zA-Z]*$", a) for a in args)
|
||||
pos = [a for a in args if not a.startswith("-")]
|
||||
refspecs = pos[1:] if len(pos) > 1 else []
|
||||
to_main = False
|
||||
for r in refspecs:
|
||||
r2 = r.lstrip("+")
|
||||
dst = r2.split(":")[-1]
|
||||
if dst in ("main", "refs/heads/main"):
|
||||
to_main = True
|
||||
if r.startswith("+") and dst in ("main", "refs/heads/main"):
|
||||
force = True
|
||||
if "--delete" in args or "-d" in args:
|
||||
if any(r in ("main", "refs/heads/main") for r in refspecs):
|
||||
return deny("main 브랜치 삭제 푸시 금지.")
|
||||
if not refspecs: # git push / git push origin → 현재 브랜치
|
||||
_, br = git(["rev-parse", "--abbrev-ref", "HEAD"], cwd)
|
||||
if br.strip() == "main":
|
||||
to_main = True
|
||||
if force and to_main:
|
||||
return deny("main 강제 푸시(--force/-f/+ref) 금지.")
|
||||
if to_main:
|
||||
return ask(deploy_reason("main 푸시 = 운영 배포(Jenkins 재기동)"))
|
||||
return None
|
||||
|
||||
|
||||
def deploy_reason(prefix: str) -> str:
|
||||
now = kst_now()
|
||||
st = market_state(now)
|
||||
msg = f"{prefix}. 현재 KST {now:%Y-%m-%d %H:%M} ({'월화수목금토일'[now.weekday()]}) {st}."
|
||||
if st == "장중":
|
||||
msg += " ⚠ 장중 배포 — 재기동이 매매에 영향(DT 포지션·스케줄러 상태 유실, 토큰 재발급)."
|
||||
return msg
|
||||
|
||||
|
||||
def check_deploy_script(argv: list[str]) -> dict | None:
|
||||
if any(t.endswith("deploy.sh") for t in argv[:3]) and "--watch-only" not in argv:
|
||||
return ask(deploy_reason("deploy.sh 실행(원격 main 푸시 포함)"))
|
||||
return None
|
||||
|
||||
|
||||
def _resolve(path: str, cwd: str) -> str:
|
||||
p = os.path.expandvars(os.path.expanduser(path))
|
||||
if p.startswith("$"):
|
||||
p = cwd
|
||||
return os.path.realpath(p if os.path.isabs(p) else os.path.join(cwd, p))
|
||||
|
||||
|
||||
def check_rm(argv: list[str], cwd: str, root: str) -> dict | None:
|
||||
if base(argv[0]) != "rm":
|
||||
return None
|
||||
flags = [a for a in argv[1:] if a.startswith("-")]
|
||||
recursive = any(a in ("--recursive",) or re.match(r"^-[a-zA-Z]*[rR][a-zA-Z]*$", a) for a in flags)
|
||||
if not recursive:
|
||||
return None
|
||||
protected = {os.path.realpath(root), os.path.realpath(os.path.join(root, "backend")),
|
||||
os.path.realpath(os.path.join(root, ".git"))}
|
||||
for t in argv[1:]:
|
||||
if t.startswith("-"):
|
||||
continue
|
||||
has_glob = any(ch in os.path.basename(t) for ch in "*?[")
|
||||
target = _resolve(os.path.dirname(t) or ".", cwd) if has_glob else _resolve(t, cwd)
|
||||
anc = os.path.realpath(root).startswith(target.rstrip("/") + "/") or target == "/"
|
||||
if target in protected or anc:
|
||||
return deny(f"rm -rf 대상 '{t}' 가 레포 루트/backend/.git(또는 그 상위)이다. 삭제 금지.")
|
||||
return None
|
||||
|
||||
|
||||
def commit_gate(cmd_segments: list[tuple[list[str], int]], cwd: str) -> dict | None:
|
||||
commit_idx = None
|
||||
for idx, (argv, _) in enumerate(cmd_segments):
|
||||
if base(argv[0]) == "git" and "commit" in argv[1:]:
|
||||
commit_idx = idx
|
||||
break
|
||||
if commit_idx is None:
|
||||
return None
|
||||
argv = cmd_segments[commit_idx][0]
|
||||
rc, top = git(["rev-parse", "--show-toplevel"], cwd)
|
||||
if rc != 0:
|
||||
return None
|
||||
root = top.strip()
|
||||
preceded_by_add = any(base(a[0]) == "git" and "add" in a[1:] for a, _ in cmd_segments[:commit_idx])
|
||||
commit_all = any(
|
||||
a == "--all" or (a.startswith("-") and not a.startswith("--") and "a" in a[1:])
|
||||
for a in argv[argv.index("commit") + 1:]
|
||||
)
|
||||
if preceded_by_add or commit_all:
|
||||
_, out = git(["status", "--porcelain", "--", "backend"], root)
|
||||
touched = bool(out.strip())
|
||||
else:
|
||||
_, out = git(["diff", "--cached", "--name-only"], root)
|
||||
touched = any(l.startswith("backend/") for l in out.splitlines())
|
||||
if not touched:
|
||||
return None
|
||||
cmd = os.environ.get("HARNESS_PYTEST_CMD")
|
||||
if not cmd:
|
||||
if not os.path.exists(os.path.join(root, ".venv", "bin", "python")):
|
||||
sys.stderr.write("[bash_guard] .venv 없음 — 커밋 게이트(유닛 스위트) 생략\n")
|
||||
return {"decision": "warn", "reason": ".venv 가 없어 커밋 전 유닛 스위트를 건너뛰었다. 직접 검증하라."}
|
||||
cmd = PYTEST_CMD_DEFAULT
|
||||
try:
|
||||
p = subprocess.run(["bash", "-c", cmd], cwd=root, capture_output=True, text=True, timeout=120)
|
||||
except subprocess.TimeoutExpired:
|
||||
return deny("커밋 게이트: 유닛 스위트가 120초 내 끝나지 않았다(실 KIS 호출 의심, CLAUDE.md §4.1). 커밋 보류.")
|
||||
except Exception as e: # noqa: BLE001
|
||||
sys.stderr.write(f"[bash_guard] 게이트 실행 오류 {e!r} — 통과\n")
|
||||
return None
|
||||
if p.returncode != 0:
|
||||
tail = "\n".join((p.stdout + "\n" + p.stderr).strip().splitlines()[-25:])
|
||||
return deny(f"커밋 게이트: backend 변경이 스테이징돼 있는데 유닛 스위트 실패(rc={p.returncode}). 고친 뒤 다시 커밋.\n--- tail ---\n{tail}")
|
||||
return None
|
||||
|
||||
|
||||
# ───────────────────────── 메인 ─────────────────────────
|
||||
def evaluate(cmd: str, cwd: str, root: str) -> dict | None:
|
||||
cmd = _strip_heredocs(cmd)
|
||||
results: list[dict] = []
|
||||
env_acc: dict = {}
|
||||
parsed: list[tuple[list[str], int]] = []
|
||||
|
||||
def handle(argv_all: list[str], depth: int = 0) -> None:
|
||||
env, argv = split_assignments(argv_all)
|
||||
if not argv:
|
||||
env_acc.update(env) # 'A=b' 단독 또는 export A=b
|
||||
return
|
||||
eff = {**env_acc, **env}
|
||||
if base(argv[0]) in SHELLS and "-c" in argv[1:] and depth < 3:
|
||||
i = argv.index("-c")
|
||||
if i + 1 < len(argv):
|
||||
for s in split_segments(argv[i + 1]):
|
||||
handle(tokenize(s), depth + 1)
|
||||
return
|
||||
if base(argv[0]) == "ssh":
|
||||
return # 원격 실행은 로컬 가드 대상 아님
|
||||
parsed.append((argv, depth))
|
||||
for fn in (check_local_stack, ):
|
||||
r = fn(argv)
|
||||
if r:
|
||||
results.append(r)
|
||||
for r in (check_alembic(argv, eff), check_pytest(argv, eff), check_git_push(argv, cwd),
|
||||
check_rm(argv, cwd, root), check_deploy_script(argv)):
|
||||
if r:
|
||||
results.append(r)
|
||||
|
||||
for seg in split_segments(cmd):
|
||||
handle(tokenize(seg))
|
||||
|
||||
g = commit_gate(parsed, cwd)
|
||||
if g:
|
||||
results.append(g)
|
||||
for kind in ("deny", "ask", "warn"):
|
||||
hit = [r for r in results if r["decision"] == kind]
|
||||
if hit:
|
||||
return {"decision": kind, "reason": " | ".join(r["reason"] for r in hit)}
|
||||
return None
|
||||
|
||||
|
||||
def main() -> int:
|
||||
data = read_stdin_json()
|
||||
if data.get("tool_name") not in (None, "Bash"):
|
||||
return 0
|
||||
cmd = (data.get("tool_input") or {}).get("command")
|
||||
if not isinstance(cmd, str) or not cmd.strip():
|
||||
return 0
|
||||
cwd = data.get("cwd") or os.getcwd()
|
||||
try:
|
||||
res = evaluate(cmd, cwd, project_dir(data))
|
||||
except Exception as e: # noqa: BLE001 훅 자체 오류는 통과
|
||||
sys.stderr.write(f"[bash_guard] internal error {e!r}\n")
|
||||
return 0
|
||||
if not res:
|
||||
return 0
|
||||
if res["decision"] == "warn":
|
||||
json.dump({"hookSpecificOutput": {"hookEventName": "PreToolUse", "additionalContext": res["reason"]}}, sys.stdout, ensure_ascii=False)
|
||||
return 0
|
||||
json.dump(
|
||||
{"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": res["decision"],
|
||||
"permissionDecisionReason": f"[harness] {res['reason']}"}},
|
||||
sys.stdout, ensure_ascii=False,
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user