vince-winkintel__gitlab-cli-skills/scripts/post-inline-comment.py

#!/usr/bin/env python3
"""
post-inline-comment.py — Post inline diff comments on GitLab MRs via JSON body.

WHY THIS SCRIPT EXISTS:
  `glab api --field position[new_line]=N` silently falls back to a general (non-inline)
  comment when GitLab rejects the position. This happens reliably for:
    - Entirely new files (new_file=True in the diff)
    - Deeply nested URL-encoded paths
    - Any case where form-encoded position fields are not parsed correctly server-side

  The fix is to send position data as a proper JSON body. This script does that.

USAGE:
  # Post a single inline comment
  python3 post-inline-comment.py \
    --project "mygroup/myproject" \
    --mr 42 \
    --file "src/utils/helpers.ts" \
    --line 16 \
    --body "This returns the wrapper object, not the value. Use .data instead."

  # Post from a JSON file (batch mode)
  python3 post-inline-comment.py --project "mygroup/myproject" --mr 42 --batch comments.json

BATCH FILE FORMAT (comments.json):
  [
    {
      "file": "src/utils/helpers.ts",
      "line": 16,
      "body": "Comment text here"
    },
    {
      "file": "src/routes/+page.svelte",
      "line": 58,
      "body": "Another comment"
    }
  ]

ENVIRONMENT:
  GITLAB_TOKEN   — Personal access token with api scope (or set via glab auth login)
  GITLAB_HOST    — GitLab host URL (default: https://gitlab.com)

SECURITY:
  - Token is read from GITLAB_TOKEN env var or glab config; never logged or echoed
  - Only HTTPS hosts are accepted (enforced at startup)
  - Comment body is capped at 10,000 characters
  - --project and --file inputs are validated before use
  - Batch files are size-limited (max 100 comments per run)
  - Token value is validated as non-empty before any API call

REQUIREMENTS:
  - Python 3.6+ (stdlib only, no pip installs needed)
  - glab authenticated (token auto-read from glab config if GITLAB_TOKEN not set)
"""

import argparse
import json
import os
import re
import ssl
import subprocess
import sys
import urllib.parse
import urllib.request
import urllib.error

# ── Security constants ────────────────────────────────────────────────────────
MAX_BODY_LENGTH = 10_000      # GitLab's own limit is ~1MB but we cap for safety
MAX_BATCH_SIZE  = 100         # prevent runaway API usage
MAX_BATCH_FILE_BYTES = 1_048_576  # 1 MB batch file limit
VALID_PROJECT_RE = re.compile(r'^[\w.\-]+(/[\w.\-]+)+$')  # group/project or group/sub/project
VALID_FILE_RE    = re.compile(r'^[^\x00\n\r]+$')          # no null bytes or newlines


# ── Token handling ────────────────────────────────────────────────────────────

def get_token(host):
    """
    Get GitLab token from env or glab config.
    Never prints or logs the token value.
    """
    token = os.environ.get("GITLAB_TOKEN", "").strip()
    if token:
        _validate_token(token)
        return token

    # Derive hostname for glab config lookup
    hostname = urllib.parse.urlparse(host).hostname or "gitlab.com"
    try:
        result = subprocess.run(
            ["glab", "config", "get", "token", "--host", hostname],
            capture_output=True, text=True, timeout=10
        )
        if result.returncode == 0:
            token = result.stdout.strip()
            if token:
                _validate_token(token)
                return token
    except (FileNotFoundError, subprocess.TimeoutExpired):
        pass

    print(
        "ERROR: No GitLab token found.\n"
        "  Set the GITLAB_TOKEN environment variable, or run: glab auth login",
        file=sys.stderr
    )
    sys.exit(1)


def _validate_token(token):
    """Basic sanity check — token must look like a PAT (non-empty, no whitespace)."""
    if not token or len(token) < 10 or re.search(r'\s', token):
        print("ERROR: GITLAB_TOKEN appears invalid (too short or contains whitespace).", file=sys.stderr)
        sys.exit(1)


# ── Input validation ──────────────────────────────────────────────────────────

def validate_host(host):
    """Enforce HTTPS to prevent token leakage over plaintext."""
    parsed = urllib.parse.urlparse(host)
    if parsed.scheme != "https":
        print(
            f"ERROR: --host must use HTTPS (got '{parsed.scheme}://').\n"
            "  Token transmission over HTTP is not allowed.",
            file=sys.stderr
        )
        sys.exit(1)
    return host.rstrip("/")


def validate_project(project):
    """Validate project path format: group/project or group/subgroup/project."""
    if not VALID_PROJECT_RE.match(project):
        print(
            f"ERROR: --project '{project}' is not a valid GitLab project path.\n"
            "  Expected format: 'group/project' or 'group/subgroup/project'",
            file=sys.stderr
        )
        sys.exit(1)
    return project


def validate_file_path(file_path):
    """Validate that a file path doesn't contain dangerous characters."""
    if not file_path or not VALID_FILE_RE.match(file_path):
        print(f"ERROR: Invalid file path: {repr(file_path)}", file=sys.stderr)
        sys.exit(1)
    return file_path


def validate_body(body):
    """Trim and cap comment body length."""
    body = body.strip()
    if not body:
        print("ERROR: Comment body cannot be empty.", file=sys.stderr)
        sys.exit(1)
    if len(body) > MAX_BODY_LENGTH:
        print(
            f"WARNING: Comment body truncated from {len(body)} to {MAX_BODY_LENGTH} characters.",
            file=sys.stderr
        )
        body = body[:MAX_BODY_LENGTH]
    return body


def validate_line(line):
    """Line number must be a positive integer."""
    if not isinstance(line, int) or line < 1:
        print(f"ERROR: Line number must be a positive integer (got {line!r}).", file=sys.stderr)
        sys.exit(1)
    return line


def load_batch_file(path):
    """Load and validate a batch comments JSON file."""
    try:
        size = os.path.getsize(path)
        if size > MAX_BATCH_FILE_BYTES:
            print(
                f"ERROR: Batch file is too large ({size} bytes, max {MAX_BATCH_FILE_BYTES}).",
                file=sys.stderr
            )
            sys.exit(1)
        with open(path) as f:
            comments = json.load(f)
    except (OSError, json.JSONDecodeError) as e:
        print(f"ERROR: Could not load batch file '{path}': {e}", file=sys.stderr)
        sys.exit(1)

    if not isinstance(comments, list):
        print("ERROR: Batch file must contain a JSON array.", file=sys.stderr)
        sys.exit(1)
    if len(comments) > MAX_BATCH_SIZE:
        print(
            f"ERROR: Batch file contains {len(comments)} comments (max {MAX_BATCH_SIZE}).",
            file=sys.stderr
        )
        sys.exit(1)

    # Validate each entry
    validated = []
    for i, c in enumerate(comments):
        if not isinstance(c, dict) or not all(k in c for k in ("file", "line", "body")):
            print(f"ERROR: Batch entry {i} missing required keys (file, line, body).", file=sys.stderr)
            sys.exit(1)
        validated.append({
            "file": validate_file_path(c["file"]),
            "line": validate_line(int(c["line"])),
            "body": validate_body(c["body"]),
        })
    return validated


# ── GitLab API helpers ────────────────────────────────────────────────────────

def _make_ssl_context():
    """Return a strict SSL context (system CA bundle, no hostname bypass)."""
    ctx = ssl.create_default_context()
    return ctx


def _api_get(token, url):
    """Authenticated GET request, returns parsed JSON."""
    req = urllib.request.Request(url, headers={"PRIVATE-TOKEN": token})
    with urllib.request.urlopen(req, context=_make_ssl_context()) as resp:
        return json.loads(resp.read())


def get_mr_versions(token, host, project_id, mr_iid):
    """Fetch current HEAD/START/BASE SHAs for an MR."""
    url = f"{host}/api/v4/projects/{project_id}/merge_requests/{mr_iid}/versions"
    versions = _api_get(token, url)
    if not versions:
        raise ValueError(f"No versions found for MR !{mr_iid}")
    latest = versions[0]
    return {
        "head_sha":  latest["head_commit_sha"],
        "start_sha": latest["start_commit_sha"],
        "base_sha":  latest["base_commit_sha"],
    }


def post_inline_comment(token, host, project_id, mr_iid, shas, file_path, line_number, body):
    """
    Post a single inline comment on a MR diff using a JSON body.
    Returns (disc_id, is_inline) tuple.
    is_inline=False means GitLab rejected the position and fell back to a general comment.
    """
    url = f"{host}/api/v4/projects/{project_id}/merge_requests/{mr_iid}/discussions"

    payload = {
        "body": body,
        "position": {
            "base_sha":      shas["base_sha"],
            "start_sha":     shas["start_sha"],
            "head_sha":      shas["head_sha"],
            "position_type": "text",
            "new_path":      file_path,
            "new_line":      line_number,
            "old_path":      file_path,  # same as new_path for added/new files
            "old_line":      None,       # None = added line (no old-side anchor)
        }
    }

    data = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        url,
        data=data,
        headers={
            "PRIVATE-TOKEN": token,
            "Content-Type":  "application/json",
        },
        method="POST"
    )

    try:
        with urllib.request.urlopen(req, context=_make_ssl_context()) as resp:
            r = json.loads(resp.read())
    except urllib.error.HTTPError as e:
        error_body = e.read().decode(errors="replace")
        raise RuntimeError(f"HTTP {e.code}: {error_body[:500]}")

    note     = r.get("notes", [{}])[0]
    disc_id  = r.get("id")
    is_inline = note.get("position") is not None

    return disc_id, is_inline


# ── Main ──────────────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(
        description="Post inline diff comments on GitLab MRs via JSON body.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--project", required=True,
                        help="GitLab project path, e.g. mygroup/myproject")
    parser.add_argument("--mr", required=True, type=int,
                        help="MR IID (integer), e.g. 42")
    parser.add_argument("--host", default="https://gitlab.com",
                        help="GitLab host URL (must be HTTPS)")
    parser.add_argument("--file",  help="File path in repo (single comment mode)")
    parser.add_argument("--line",  type=int, help="Line number in new file (single comment mode)")
    parser.add_argument("--body",  help="Comment text (single comment mode)")
    parser.add_argument("--batch", help="Path to JSON file with [{file, line, body}] array")
    args = parser.parse_args()

    if not args.batch and not (args.file and args.line and args.body):
        parser.error("Provide either --batch or all of --file, --line, --body")

    # Validate all inputs before touching the network
    host       = validate_host(args.host)
    project    = validate_project(args.project)
    project_id = urllib.parse.quote(project, safe="")

    if args.batch:
        comments = load_batch_file(args.batch)
    else:
        comments = [{
            "file": validate_file_path(args.file),
            "line": validate_line(args.line),
            "body": validate_body(args.body),
        }]

    # Fetch token after validation (avoids unnecessary credential access on bad input)
    token = get_token(host)

    print(f"Fetching current HEAD SHAs for MR !{args.mr}...")
    try:
        shas = get_mr_versions(token, host, project_id, args.mr)
    except Exception as e:
        print(f"ERROR: Could not fetch MR versions: {e}", file=sys.stderr)
        sys.exit(1)
    print(f"  head_sha: {shas['head_sha'][:12]}...")

    results = []
    for c in comments:
        file_path   = c["file"]
        line_number = c["line"]
        body        = c["body"]

        print(f"\nPosting: {file_path}:{line_number}")
        print(f"  Body: {body[:80]}{'...' if len(body) > 80 else ''}")

        try:
            disc_id, is_inline = post_inline_comment(
                token, host, project_id, args.mr, shas, file_path, line_number, body
            )
            status = "✅ INLINE" if is_inline else "⚠️  GENERAL (position rejected — check line number)"
            print(f"  {status} | disc_id: {disc_id}")
            results.append({
                "disc_id":   disc_id,
                "is_inline": is_inline,
                "file":      file_path,
                "line":      line_number,
            })
        except Exception as e:
            print(f"  ❌ FAILED: {e}", file=sys.stderr)
            results.append({"error": str(e), "file": file_path, "line": line_number})

    # Summary
    print(f"\n{'=' * 50}")
    inline_count  = sum(1 for r in results if r.get("is_inline") is True)
    general_count = sum(1 for r in results if r.get("is_inline") is False)
    error_count   = sum(1 for r in results if "error" in r)
    print(f"Summary: {inline_count} inline ✅  {general_count} general ⚠️  {error_count} failed ❌")

    if general_count:
        print(
            "\n⚠️  Some comments posted as general (non-inline).\n"
            "   The line number doesn't correspond to an added (+) line in the diff.\n"
            "   Use the get_new_line_number() helper in glab-mr/SKILL.md to find valid lines."
        )

    disc_ids = [r["disc_id"] for r in results if r.get("disc_id")]
    if disc_ids:
        print(f"\nDiscussion IDs: {json.dumps(disc_ids)}")

    return 0 if error_count == 0 else 1


if __name__ == "__main__":
    sys.exit(main())