#!/usr/bin/env python3 """ giip_issue.py -- GIIP Issue REST API client (giip skill "giip-issue", v1.0.0) Standard-library only (urllib.request). No third-party dependencies, so it runs on hosts where installing packages is uncertain or disallowed. Safety contract (see SKILL.md for the authoritative copy): - Base URL: https://giipfaw.azurewebsites.net/api - Auth: `x-api-key` header ONLY. The key is read from the GIIP_API_KEY environment variable at run time and is never written to disk, logs, or exception messages. - `create` performs a read-only preflight (GET /giipIssues?csn=...) before writing. If the preflight returns HTTP 200 with an empty `issues` array, permission on the target csn is unconfirmed and the create is aborted (exit code 4) -- no write is attempted. - After `create` succeeds, the script re-fetches the issue by the returned isn (GET /giipIssues?isn=...) and compares csn/title/status against what was requested. A csn permission mismatch is NOT always a 401: it can come back as HTTP 200 with an empty issues array (read) or as a silent clamp to the key's home csn (POST). Any mismatch is reported (exit code 5) -- the script never auto-retries or auto-corrects. - `update-status` sends the minimal PUT payload {"isn":..., "status":...} only. It never includes title/content/csn or any other field, because the underlying endpoint performs a partial (ISNULL-preserving) update and any field present in the JSON body overwrites the stored value. - `comment` posts to POST /giipIssueComments. `author` and `issuetype` default to "giip-issue-skill" and "comment" respectively when omitted. The server may normalize the stored `author` to the caller's actual identity -- callers should not assume the value they sent is what was persisted. - There is no delete subcommand. The GIIP Issue API does not expose one. - On HTTP error, JSON parse failure, or timeout, the script reports a distinct exit code and a message that never includes the API key value. It never automatically retries a write. Exit codes: 0 success 1 usage / argument error 2 GIIP_API_KEY environment variable is not set 3 HTTP / network error (includes timeout) talking to the API 4 create: csn preflight returned no visible issues (permission unconfirmed) 5 create: post-write verification mismatch (csn/title/status did not match) 6 JSON parse error in the API response """ import argparse import json import os import sys import urllib.error import urllib.request BASE_URL = "https://giipfaw.azurewebsites.net/api" TIMEOUT_SECONDS = 15 DEFAULT_COMMENT_AUTHOR = "giip-issue-skill" DEFAULT_COMMENT_ISSUETYPE = "comment" DEFAULT_CREATE_STATUS = "PENDING" def _get_api_key(): """Read the API key from the environment. Never log or echo the value.""" key = os.environ.get("GIIP_API_KEY") if not key: _emit_and_exit( success=False, error="GIIP_API_KEY environment variable is not set. " "Set it for this process only; do not persist it to a file.", exit_code=2, ) return key def _emit_and_exit(success, exit_code, error=None, **fields): """Print the standard JSON result shape to stdout and exit.""" payload = {"success": success} payload.update(fields) if error is not None: payload["error"] = error print(json.dumps(payload, ensure_ascii=False)) sys.exit(exit_code) def _request(method, path, api_key, query=None, body=None): """ Issue one HTTP request. Returns (status_code, parsed_json_or_None, raw_text). Never includes api_key in any exception message. """ url = BASE_URL + path if query: qs = "&".join( "{}={}".format(k, urllib.request.quote(str(v))) for k, v in query.items() if v is not None ) if qs: url = url + "?" + qs data = None headers = {"x-api-key": api_key} if body is not None: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS) as resp: raw = resp.read().decode("utf-8", errors="replace") status = resp.getcode() except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", errors="replace") if e.fp else "" status = e.code except urllib.error.URLError as e: _emit_and_exit( success=False, exit_code=3, error="Network error contacting GIIP API: {}".format(e.reason), ) except TimeoutError: _emit_and_exit( success=False, exit_code=3, error="Timed out after {}s contacting GIIP API. " "Do not auto-retry; verify manually (list/get) before writing.".format( TIMEOUT_SECONDS ), ) parsed = None if raw: try: parsed = json.loads(raw) except (ValueError, json.JSONDecodeError): _emit_and_exit( success=False, exit_code=6, error="Could not parse GIIP API response as JSON (HTTP {}).".format( status ), ) return status, parsed def cmd_list(args): api_key = _get_api_key() query = {"csn": args.csn} if args.status: query["status"] = args.status status, parsed = _request("GET", "/giipIssues", api_key, query=query) if status != 200: _emit_and_exit( success=False, exit_code=3, error="GET /giipIssues failed with HTTP {}.".format(status), ) issues = (parsed or {}).get("issues", []) _emit_and_exit(success=True, exit_code=0, csn=args.csn, issues=issues) def cmd_get(args): api_key = _get_api_key() status, parsed = _request("GET", "/giipIssues", api_key, query={"isn": args.isn}) if status != 200: _emit_and_exit( success=False, exit_code=3, error="GET /giipIssues failed with HTTP {}.".format(status), ) issues = (parsed or {}).get("issues", []) if not issues: _emit_and_exit( success=False, exit_code=3, isn=args.isn, error="No issue returned for isn={} (not found, or no permission " "on its csn).".format(args.isn), ) issue = issues[0] _emit_and_exit( success=True, exit_code=0, isn=args.isn, csn=issue.get("cSn", issue.get("csn")), status=issue.get("status"), title=issue.get("title"), issue=issue, ) def cmd_create(args): api_key = _get_api_key() # 1) Read-only preflight: confirm read access to the target csn before # attempting any write. pre_status, pre_parsed = _request( "GET", "/giipIssues", api_key, query={"csn": args.csn} ) if pre_status != 200: _emit_and_exit( success=False, exit_code=3, error="Preflight GET /giipIssues?csn={} failed with HTTP {}.".format( args.csn, pre_status ), ) pre_issues = (pre_parsed or {}).get("issues", []) if not pre_issues: _emit_and_exit( success=False, exit_code=4, csn=args.csn, error="Permission on csn={} could not be confirmed (preflight " "returned issues: []). Refusing to create -- verify the csn or " "ask an administrator for access.".format(args.csn), ) # 2) Create. body = { "title": args.title, "content": args.content, "status": args.status or DEFAULT_CREATE_STATUS, "csn": args.csn, "target_lssn": args.target_lssn, "agent_workflow": args.agent_workflow, } status, parsed = _request("POST", "/giipIssues", api_key, body=body) if status != 200 or not parsed or not parsed.get("success"): _emit_and_exit( success=False, exit_code=3, error="POST /giipIssues failed with HTTP {}. Do not auto-retry; " "check list/get for a possible existing registration before " "trying again by hand.".format(status), ) new_isn = parsed.get("isn") # 3) Re-fetch by the returned isn and verify csn/title/status. The # target csn can be silently clamped to the key's home csn on # permission mismatch, so this step is mandatory, not optional. v_status, v_parsed = _request( "GET", "/giipIssues", api_key, query={"isn": new_isn} ) if v_status != 200: _emit_and_exit( success=False, exit_code=5, isn=new_isn, error="Issue {} was created but post-write verification GET " "failed with HTTP {}. Verify manually.".format(new_isn, v_status), ) v_issues = (v_parsed or {}).get("issues", []) if not v_issues: _emit_and_exit( success=False, exit_code=5, isn=new_isn, error="Issue {} was created but verification GET returned no " "issue. Verify manually before assuming success.".format(new_isn), ) actual = v_issues[0] actual_csn = actual.get("cSn", actual.get("csn")) actual_title = actual.get("title") actual_status = actual.get("status") mismatches = [] if actual_csn != args.csn: mismatches.append( "csn requested={} actual={} (likely silently clamped to the " "key's home csn)".format(args.csn, actual_csn) ) if actual_title != args.title: mismatches.append("title requested={!r} actual={!r}".format(args.title, actual_title)) expected_status = args.status or DEFAULT_CREATE_STATUS if actual_status != expected_status: mismatches.append( "status requested={!r} actual={!r}".format(expected_status, actual_status) ) if mismatches: _emit_and_exit( success=False, exit_code=5, isn=new_isn, csn=actual_csn, status=actual_status, title=actual_title, error="Issue {} was created but does not match what was " "requested: {}. Not auto-correcting -- report this and resolve " "manually.".format(new_isn, "; ".join(mismatches)), ) _emit_and_exit( success=True, exit_code=0, isn=new_isn, csn=actual_csn, status=actual_status, title=actual_title, ) def cmd_update_status(args): api_key = _get_api_key() # Minimal payload only -- never include title/content/csn here, since # the endpoint is a partial update and any field present overwrites # the stored value. body = {"isn": args.isn, "status": args.status} status, parsed = _request("PUT", "/giipIssues", api_key, body=body) if status != 200 or not parsed or not parsed.get("success"): _emit_and_exit( success=False, exit_code=3, isn=args.isn, error="PUT /giipIssues failed with HTTP {}.".format(status), ) _emit_and_exit(success=True, exit_code=0, isn=args.isn, status=args.status) def cmd_comment(args): api_key = _get_api_key() body = { "isn": args.isn, "content": args.content, "author": args.author or DEFAULT_COMMENT_AUTHOR, "issuetype": args.issuetype or DEFAULT_COMMENT_ISSUETYPE, } status, parsed = _request("POST", "/giipIssueComments", api_key, body=body) if status != 200 or not parsed or not parsed.get("success"): _emit_and_exit( success=False, exit_code=3, isn=args.isn, error="POST /giipIssueComments failed with HTTP {}.".format(status), ) _emit_and_exit(success=True, exit_code=0, isn=args.isn) def build_parser(): p = argparse.ArgumentParser( prog="giip_issue.py", description="GIIP Issue REST API client (list/get/create/update-status/comment). " "No delete subcommand -- the API does not provide one.", ) sub = p.add_subparsers(dest="command", required=True) p_list = sub.add_parser("list", help="List issues for a csn.") p_list.add_argument("--csn", type=int, required=True) p_list.add_argument("--status", type=str, default=None) p_list.set_defaults(func=cmd_list) p_get = sub.add_parser("get", help="Get one issue by isn.") p_get.add_argument("--isn", type=int, required=True) p_get.set_defaults(func=cmd_get) p_create = sub.add_parser( "create", help="Create an issue after a read-access preflight; verifies after write." ) p_create.add_argument("--csn", type=int, required=True) p_create.add_argument("--title", type=str, required=True) p_create.add_argument("--content", type=str, required=True) p_create.add_argument("--status", type=str, default=None) p_create.add_argument("--target-lssn", type=int, default=None) p_create.add_argument("--agent-workflow", type=str, default=None) p_create.set_defaults(func=cmd_create) p_upd = sub.add_parser( "update-status", help="Update only the status field of an issue (minimal payload)." ) p_upd.add_argument("--isn", type=int, required=True) p_upd.add_argument("--status", type=str, required=True) p_upd.set_defaults(func=cmd_update_status) p_comment = sub.add_parser("comment", help="Add a comment to an issue.") p_comment.add_argument("--isn", type=int, required=True) p_comment.add_argument("--content", type=str, required=True) p_comment.add_argument("--author", type=str, default=None) p_comment.add_argument("--issuetype", type=str, default=None) p_comment.set_defaults(func=cmd_comment) return p def main(): parser = build_parser() args = parser.parse_args() args.func(args) if __name__ == "__main__": main()