#!/usr/bin/env python3 """Shell/CLI 예제에서 교육성 검토가 필요한 신호를 결정론적으로 찾는다. 이 모듈의 finding은 금지 규칙이 아니다. planner/editor/reviewer가 볼 범위를 좁히는 신호다. 기술적으로 정당한 compact command는 문맥에 따라 그대로 남을 수 있다. """ from __future__ import annotations import hashlib import re from typing import Any SHELL_FENCE_RE = re.compile( r"^```[ \t]*(?Pbash|sh|shell|zsh|console)" r"(?P(?:[ \t]+[^\n]*)?)\n" r"(?P.*?)^```[ \t]*$", re.MULTILINE | re.DOTALL | re.IGNORECASE, ) FENCE_LABEL_RE = re.compile( r"(?:^|\s)label\s*=\s*(?:\"(?P[^\"]*)\"|'(?P[^']*)')", re.IGNORECASE, ) LABEL_MODE_RE = re.compile( r"\[(tutorial|operator|diagnostic|automation|reference)\]", re.IGNORECASE, ) LABEL_EXECUTION_CONTEXT_RE = re.compile( r"(?:\blab\s+host\b|\bkc-lab-[\w-]+\b|\btest-server\b|\bdev\b|" r"\bworkstation\b|\bhost\b|\bserver\b|\bguest\b|\bcontainer\b|\bpod\b|" r"워크스테이션|호스트|서버|게스트|컨테이너|파드)", re.IGNORECASE, ) RAW_SSH_IP_RE = re.compile( r"(?m)^\s*(?:ssh|scp)\b[^\n]*?(?P(?:\d{1,3}\.){3}\d{1,3})(?=[:\s]|$)" ) REMOTE_COMMAND_RE = re.compile(r"(?m)^\s*(?:ssh|scp)\b") EXECUTION_CONTEXT_RE = re.compile( r"(?:실행\s*위치|로컬|원격|호스트|서버|게스트|컨테이너|\bVM\b|" r"\blocal\b|\bremote\b|\bhost\b|\bserver\b|\bguest\b|" r"\bcontainer\b|\bworkstation\b)", re.IGNORECASE, ) REMOTE_COMPOUND_RE = re.compile( r"(?m)^\s*ssh\s+[^\n]+?(['\"])(?:(?!\1).)*(?:;|&&|\|\|)(?:(?!\1).)*\1" ) CLEANUP_CHAIN_RE = re.compile( r"(?m)^.*(?:schema|verify|check|test|validate)[^\n]*(?:;|&&)\s*rm\b", re.IGNORECASE, ) PIPE_OPERATOR_RE = re.compile(r"(?:^|\s)\|(?!\|)\s+", re.MULTILINE) TEXT_FENCE_RE = re.compile( r"^```text\s*\n(?P.*?)^```[ \t]*$", re.MULTILINE | re.DOTALL | re.IGNORECASE, ) COMMAND_MODE_MARKER_RE = re.compile( r"", re.IGNORECASE, ) COMMAND_MODE_MARKER_BEFORE_BLOCK_RE = re.compile( r"\s*$", re.IGNORECASE, ) COMMAND_LIKE_LINE_RE = re.compile( r"(?i)(?:\./gradlew|\b(?:docker|kubectl|git|npm|pnpm|yarn|npx|python3?|node|" r"gradle|ssh|scp|curl|helm|virsh|systemctl|journalctl)\b)" ) COMMAND_MODES = {"tutorial", "operator", "diagnostic", "automation", "reference"} COMPACT_ALLOWED_MODES = {"automation", "reference"} def sha256_text(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest() def _finding( *, block_id: str, code: str, severity: str, evidence: str, instruction: str ) -> dict[str, Any]: return { "block_id": block_id, "code": code, "severity": severity, "evidence": evidence[:240], "instruction": instruction, } def _block_findings(block_id: str, body: str, mode: str) -> list[dict[str, Any]]: findings: list[dict[str, Any]] = [] compact_allowed = mode in COMPACT_ALLOWED_MODES if not compact_allowed and re.search( r"(?m)^\s*sed\s+(?:-[A-Za-z]*i[A-Za-z]*|--in-place(?:=[^\s]+)?)(?:\s|$)", body, ): findings.append( _finding( block_id=block_id, code="in-place-text-rewrite", severity="minor", evidence="sed -i", instruction=( "튜토리얼/운영 흐름에서는 명시적 편집 단계를 우선하고, 자동화 경계라면 " "in-place rewrite가 필요한 이유를 설명한다." ), ) ) if not compact_allowed and "$(" in body: findings.append( _finding( block_id=block_id, code="command-substitution", severity="minor", evidence="$(...)", instruction=( "값을 먼저 확인하거나 명시적으로 입력할 수 있으면 command substitution을 " "첫 학습 경로로 두지 않는다." ), ) ) if not compact_allowed and re.search(r"\$\([^)]*\$\(", body, re.DOTALL): findings.append( _finding( block_id=block_id, code="nested-command-substitution", severity="minor", evidence="$(...$(...)...)", instruction=( "중첩된 command substitution은 값을 단계별로 확인하거나 별도 스크립트로 분리한다." ), ) ) raw_ssh = RAW_SSH_IP_RE.search(body) if raw_ssh: findings.append( _finding( block_id=block_id, code="raw-ssh-ip", severity="minor", evidence=raw_ssh.group(0).strip(), instruction=( "같은 관리 호스트를 반복해서 쓴다면 SSH alias를 먼저 정의하고 주 흐름에서는 " "alias를 사용한다." ), ) ) compound = REMOTE_COMPOUND_RE.search(body) if compound: findings.append( _finding( block_id=block_id, code="compound-remote-shell", severity="minor", evidence=compound.group(0).strip(), instruction=( "복합 원격 셸은 검토 신호다. 읽기 전용 확인이나 검증 성공 뒤 변경하는 " "안전 조건이면 유지할 수 있고, 서로 다른 효과를 숨기면 관찰 가능한 단계로 분리한다." ), ) ) cleanup = CLEANUP_CHAIN_RE.search(body) if cleanup: findings.append( _finding( block_id=block_id, code="cleanup-chained-with-verification", severity="minor" if mode == "reference" else "major", evidence=cleanup.group(0).strip(), instruction=( "reference block이면 위험한 historical form을 그대로 보존하되 따라 하는 절차와 " "분리해 설명한다. 실행 절차라면 검증과 파괴적 정리를 별도 명령으로 분리해 " "삭제 전에 실패를 관찰할 수 있게 한다." ), ) ) if not compact_allowed and re.search(r"(?m)^\s*printf\b[^\n]*>\s*[^\s]+", body): findings.append( _finding( block_id=block_id, code="opaque-file-generation", severity="minor", evidence="printf ... > file", instruction=( "작은 설정/학습 파일은 생성 자동화보다 파일 내용을 먼저 직접 보여준다." ), ) ) if re.search(r"(?:2>|&>)\s*/dev/null|2>&1", body): findings.append( _finding( block_id=block_id, code="hidden-stderr", severity="minor", evidence="stderr redirection", instruction=( "주 진단 경로에서 stderr를 숨기지 말고, 원래 출력을 본 뒤 필요한 경우에만 " "필터링한다." ), ) ) if not compact_allowed and len(PIPE_OPERATOR_RE.findall(body)) >= 2: findings.append( _finding( block_id=block_id, code="compressed-pipeline", severity="minor", evidence="multi-stage shell pipeline", instruction=( "중간 출력이 학습에 필요하면 파이프라인을 나누거나 각 단계를 설명한 뒤 " "compact form을 제시한다." ), ) ) if not compact_allowed: cd_match = re.search(r"(?m)^\s*cd\s+\S+.*$", body) if cd_match and re.search(r"(?m)^\s*#\s*\S.*$", body[cd_match.end():]): findings.append( _finding( block_id=block_id, code="cwd-transition-crosses-command-group", severity="minor", evidence=cd_match.group(0).strip(), instruction=( "cwd를 바꾼 뒤 다른 명령 그룹이 이어지면 실행 위치가 숨는다. " "그룹을 나누거나 이후 명령의 cwd를 다시 명시한다." ), ) ) return findings def _explicit_block_mode(text: str, start: int) -> str | None: """블록과 공백만 사이에 둔 command-mode marker만 해당 블록에 결속한다.""" nearby = text[max(0, start - 320):start] marker = COMMAND_MODE_MARKER_BEFORE_BLOCK_RE.search(nearby) return marker.group(1).lower() if marker else None def _block_mode(text: str, start: int, default_mode: str) -> str: """명시 mode가 있으면 쓰고, 없으면 문서 기본 mode를 쓴다.""" return _explicit_block_mode(text, start) or default_mode def _label_mode(label: str | None) -> str | None: """Studio가 허용하는 단일 label 안의 [reference] 같은 mode marker를 읽는다.""" if not label: return None match = LABEL_MODE_RE.search(label) return match.group(1).lower() if match else None def _fence_label(info: str) -> str | None: """Markdown info string의 label 값을 반환한다. 첫 토큰(language) 뒤의 metadata는 Markdown renderer용 정보이므로 command body와 섞지 않는다. 다만 `[lab host]`처럼 실행 위치를 적은 label은 사람이 보는 문맥과 동일하게 execution-context 판정에 사용할 수 있다. """ match = FENCE_LABEL_RE.search(info) if not match: return None value = ( match.group("double") if match.group("double") is not None else match.group("single") ) value = value.strip() return value or None def _command_like_text_blocks(text: str) -> list[dict[str, Any]]: """실행 흐름처럼 보이지만 text fence로 적힌 블록을 낮은 신뢰도의 신호로 남긴다.""" out: list[dict[str, Any]] = [] for ordinal, match in enumerate(TEXT_FENCE_RE.finditer(text), start=1): body = match.group("body") explicit_mode = _explicit_block_mode(text, match.start()) if explicit_mode in {"reference", "automation"}: continue command_hits = COMMAND_LIKE_LINE_RE.findall(body) arrow_hits = len(re.findall(r"(?:→|->)", body)) if len(command_hits) < 2 and not (command_hits and arrow_hits >= 2): continue source = match.group(0) out.append( { "id": f"text-command-like-{ordinal:03d}", "start": match.start(), "end": match.end(), "source_sha256": sha256_text(source), "evidence": body[:240], "instruction": ( "개념 흐름도라면 text fence로 둬도 된다. 복사해 실행할 절차라면 shell fence와 " "실행 위치·전제조건·검증 결과를 명시한다." ), } ) return out def analyze_commands( section_id: str, text: str, *, mode: str = "operator" ) -> dict[str, Any]: """Markdown의 shell fence를 찾아 command-pedagogy 신호를 반환한다. 문서 기본 mode와 별개로 블록 직전의 command-mode marker가 있으면 그 블록만 해당 mode로 분석한다. """ if mode not in COMMAND_MODES: raise ValueError(f"unsupported command pedagogy mode: {mode}") blocks: list[dict[str, Any]] = [] findings: list[dict[str, Any]] = [] raw_hosts: dict[str, list[str]] = {} for ordinal, match in enumerate(SHELL_FENCE_RE.finditer(text), start=1): source = match.group(0) body = match.group("body") info = match.group("info") or "" label = _fence_label(info) block_id = f"command-{ordinal:03d}" explicit_mode = _label_mode(label) or _explicit_block_mode(text, match.start()) block_mode = explicit_mode or mode blocks.append( { "id": block_id, "ordinal": ordinal, "language": match.group("language").lower(), "info_string": info.strip(), "label": label, "mode": block_mode, "mode_explicit": explicit_mode is not None, "start": match.start(), "end": match.end(), "source_sha256": sha256_text(source), "body_sha256": sha256_text(body), "source": source, "body": body, } ) findings.extend(_block_findings(block_id, body, block_mode)) if REMOTE_COMMAND_RE.search(body): prose_before = text[max(0, match.start() - 320) : match.start()] label_has_context = bool( label and LABEL_EXECUTION_CONTEXT_RE.search(label) ) if not EXECUTION_CONTEXT_RE.search(prose_before) and not label_has_context: findings.append( _finding( block_id=block_id, code="execution-context-implicit", severity="minor", evidence=REMOTE_COMMAND_RE.search(body).group(0).strip(), instruction=( "원격 명령 앞에서 이 명령을 어느 host/session에서 실행하는지 명시한다." ), ) ) for remote in RAW_SSH_IP_RE.finditer(body): raw_hosts.setdefault(remote.group("ip"), []).append(block_id) for host, block_ids in sorted(raw_hosts.items()): if len(block_ids) < 2: continue findings.append( _finding( block_id=block_ids[0], code="repeated-raw-ssh-host", severity="minor", evidence=f"{host} repeated {len(block_ids)} times", instruction=( "같은 관리 호스트를 반복하면 SSH alias를 사전 설정으로 둘 수 있는지 검토한다." ), ) ) return { "schema_version": "1.0", "authority": "deterministic", "gate": "command-pedagogy-signals", "section_id": section_id, "mode": mode, "source_sha256": sha256_text(text), "result": "WARN" if findings else "PASS", "requires_editor": bool(findings), "blocks": blocks, "findings": findings, "extensions": { "command_like_text_blocks": _command_like_text_blocks(text), }, } def _require_non_empty_string(value: Any, field: str) -> str: if not isinstance(value, str) or not value.strip(): raise ValueError(f"{field} must be a non-empty string") return value def _require_sha256(value: Any, field: str) -> str: value = _require_non_empty_string(value, field) if not re.fullmatch(r"[0-9a-f]{64}", value): raise ValueError(f"{field} must be a lowercase sha256") return value def validate_command_plan( plan: dict[str, Any], *, analysis: dict[str, Any] | None = None ) -> None: """CommandPlan의 구조와 frozen analysis 결속을 검증한다.""" if not isinstance(plan, dict): raise ValueError("CommandPlan must be an object") if plan.get("schema_version") != "1.0": raise ValueError("CommandPlan schema_version must be 1.0") _require_non_empty_string(plan.get("section_id"), "CommandPlan section_id") source_sha = _require_sha256(plan.get("source_sha256"), "CommandPlan source_sha256") mode = plan.get("mode") if mode not in COMMAND_MODES: raise ValueError(f"CommandPlan mode is unsupported: {mode!r}") groups = plan.get("command_groups") if not isinstance(groups, list) or not groups: raise ValueError("CommandPlan command_groups must be a non-empty list") known_blocks: dict[str, dict[str, Any]] = {} if analysis is not None: if analysis.get("source_sha256") != source_sha: raise ValueError("CommandPlan source_sha256 does not match analysis") known_blocks = { str(block.get("id")): block for block in analysis.get("blocks", []) if isinstance(block, dict) } seen_ids: set[str] = set() seen_blocks: set[str] = set() for group in groups: if not isinstance(group, dict): raise ValueError("CommandPlan command group must be an object") group_id = _require_non_empty_string(group.get("id"), "CommandPlan command group id") if group_id in seen_ids: raise ValueError("CommandPlan contains duplicate command group id") seen_ids.add(group_id) block_id = _require_non_empty_string(group.get("block_id"), "CommandPlan block_id") if block_id in seen_blocks: raise ValueError("CommandPlan contains duplicate block_id") seen_blocks.add(block_id) block_sha = _require_sha256(group.get("source_sha256"), "CommandPlan group source_sha256") if known_blocks: block = known_blocks.get(block_id) if block is None: raise ValueError("CommandPlan references an unknown block") if block.get("source_sha256") != block_sha: raise ValueError("CommandPlan block source_sha256 does not match analysis") _require_non_empty_string(group.get("goal"), "CommandPlan goal") context = group.get("execution_context") if not isinstance(context, dict): raise ValueError("CommandPlan execution_context must be an object") _require_non_empty_string(context.get("host"), "CommandPlan execution_context.host") _require_non_empty_string(context.get("cwd"), "CommandPlan execution_context.cwd") prerequisites = group.get("prerequisites") if not isinstance(prerequisites, list) or not all( isinstance(item, str) and item.strip() for item in prerequisites ): raise ValueError("CommandPlan prerequisites must be a string list") steps = group.get("steps") if not isinstance(steps, list) or not steps: raise ValueError("CommandPlan steps must be a non-empty list") for step in steps: if not isinstance(step, dict): raise ValueError("CommandPlan step must be an object") _require_non_empty_string(step.get("command"), "CommandPlan step.command") _require_non_empty_string(step.get("reason"), "CommandPlan step.reason") _require_non_empty_string( step.get("expected_result"), "CommandPlan step.expected_result" ) cleanup = group.get("cleanup") if not isinstance(cleanup, list): raise ValueError("CommandPlan cleanup must be a list") for step in cleanup: if not isinstance(step, dict): raise ValueError("CommandPlan cleanup step must be an object") _require_non_empty_string(step.get("command"), "CommandPlan cleanup.command") _require_non_empty_string(step.get("reason"), "CommandPlan cleanup.reason") def validate_command_patch_set( patch_set: dict[str, Any], *, analysis: dict[str, Any] | None = None ) -> None: """CommandPatchSet의 구조와 frozen analysis 결속을 검증한다.""" if not isinstance(patch_set, dict): raise ValueError("CommandPatchSet must be an object") if patch_set.get("schema_version") != "1.0": raise ValueError("CommandPatchSet schema_version must be 1.0") source_sha = _require_sha256( patch_set.get("source_sha256"), "CommandPatchSet source_sha256" ) patches = patch_set.get("patches") if not isinstance(patches, list): raise ValueError("CommandPatchSet patches must be a list") known_blocks: dict[str, dict[str, Any]] = {} if analysis is not None: if analysis.get("source_sha256") != source_sha: raise ValueError("CommandPatchSet source_sha256 does not match analysis") known_blocks = { str(block.get("id")): block for block in analysis.get("blocks", []) if isinstance(block, dict) } seen: set[str] = set() for patch in patches: if not isinstance(patch, dict): raise ValueError("CommandPatchSet patch must be an object") block_id = _require_non_empty_string(patch.get("block_id"), "CommandPatchSet block_id") if block_id in seen: raise ValueError("CommandPatchSet contains duplicate block ids") seen.add(block_id) expected = _require_sha256( patch.get("expected_sha256"), "CommandPatchSet expected_sha256" ) if known_blocks: block = known_blocks.get(block_id) if block is None: raise ValueError("CommandPatchSet references an unknown block") if block.get("source_sha256") != expected: raise ValueError("CommandPatchSet expected block hash is stale") _require_non_empty_string( patch.get("replacement_markdown"), "CommandPatchSet replacement_markdown" ) def apply_command_patch_set( *, text: str, analysis: dict[str, Any], patch_set: dict[str, Any] ) -> str: """분석 당시의 shell fence span 안에서만 replacement를 적용한다. editor가 Markdown 전체를 직접 다시 쓰지 않게 하는 경계다. patch는 분석에 존재하는 block_id와 그때의 source hash를 함께 제시해야 한다. """ validate_command_patch_set(patch_set, analysis=analysis) source_sha256 = sha256_text(text) if analysis.get("source_sha256") != source_sha256: raise ValueError("command analysis source hash is stale") if patch_set.get("source_sha256") != source_sha256: raise ValueError("CommandPatchSet source hash is stale") blocks = {item["id"]: item for item in analysis.get("blocks", [])} patches = patch_set.get("patches") if not isinstance(patches, list): raise ValueError("CommandPatchSet patches must be a list") seen: set[str] = set() replacements: list[tuple[int, int, str]] = [] for patch in patches: if not isinstance(patch, dict): raise ValueError("CommandPatchSet patch must be an object") block_id = str(patch.get("block_id") or "") if block_id in seen: raise ValueError("CommandPatchSet contains duplicate block ids") seen.add(block_id) block = blocks.get(block_id) if block is None: raise ValueError("CommandPatchSet references an unknown block") expected = str(patch.get("expected_sha256") or "") if expected != block["source_sha256"]: raise ValueError("CommandPatchSet expected block hash is stale") source = text[block["start"] : block["end"]] if sha256_text(source) != block["source_sha256"]: raise ValueError("command block span is stale") replacement = patch.get("replacement_markdown") if not isinstance(replacement, str) or not replacement: raise ValueError("CommandPatchSet replacement_markdown must be non-empty") replacements.append((block["start"], block["end"], replacement)) repaired = text for start, end, replacement in sorted(replacements, reverse=True): repaired = repaired[:start] + replacement + repaired[end:] return repaired