Read it before you run it

python3 gitlab-setup.py --gitlab-url https://gitlab.example.com --dry-run
python3 gitlab-setup.py --gitlab-url https://gitlab.example.com

#!/usr/bin/env python3
"""Register the Jira integration's GitLab system hook. One run, no stored state.

This script is run by a GitLab administrator, inside their own network, against
their own self-managed GitLab. It registers ONE instance-wide system hook that
delivers push and merge request events to Jira. There is no per-repository
configuration, now or later.

    python3 gitlab-setup.py --gitlab-url https://gitlab.example.com --dry-run
    python3 gitlab-setup.py --gitlab-url https://gitlab.example.com

Run it again whenever you rotate the shared secret in Jira: it finds the hook it
created and updates it in place.

================================ SECURITY ==================================
The admin token is used for this one run and then it is gone. Concretely, and
verifiably in this file:

  * It is read from a no-echo prompt, or from stdin (--token-stdin for the
    token alone, --stdin for an unattended run). There is deliberately NO
    --token flag and no environment variable: both are readable by other
    users on this host (`ps`, /proc/<pid>/environ), and a flag also lands
    in your shell history. Feed --stdin from a file or a password manager,
    never from `echo`, which puts the secret in your history anyway.
  * It is held in one attribute, Session._token, and read by exactly one
    method, Session._call, which puts it in the PRIVATE-TOKEN header.
  * Nothing is written to disk. No config file, no cache, no state. The
    script has no write-to-file call in it at all.
  * It is never printed. Not on the success path, not in an error, not in a
    traceback -- everything printed on a failure path goes through scrub().
    There is no debug or verbose flag that dumps headers or request bodies;
    that flag is exactly how this project leaked a credential once before.
  * Redirects are refused (_NoRedirect), so the header cannot be replayed to
    a host other than the one you named.

The shared secret from the Jira config page is treated the same way, and is
only ever shown as a short fingerprint.

The web trigger URL IS printed in full, deliberately. Its path segment is a
credential, but it is also the thing you are being asked to approve, and you
cannot verify a target you cannot see. The boundary this script enforces is:
nothing on disk, nothing in the argument list, nothing in shell history,
nothing in a shared log. Your own terminal is the intended destination.
============================================================================

Exit codes: 0 success or dry run, 2 bad usage (argparse), 3 refused to act,
4 API or authentication error, 5 TLS verification failed.

Requires Python 3.6+ and nothing else. No third-party packages.
"""

import argparse
import getpass
import hashlib
import json
import ssl
import sys
import traceback
import urllib.error
import urllib.request
from urllib.parse import urlsplit, urlunsplit

EXIT_OK = 0
EXIT_REFUSED = 3
EXIT_API = 4
EXIT_TLS = 5

# The events this hook subscribes to. This set tracks what the Jira app's
# handler actually consumes today (src/index.js: object_kind "push" and
# "merge_request"); everything else it receives it discards. Subscribing to
# an event we discard is not free -- it costs one Forge invocation per
# delivery, on every tag push across every project on the instance.
#
# Every flag is sent explicitly, including the False ones. GitLab documents
# no defaults for these, and repository_update_events has historically
# defaulted to on, so an omitted flag is not the same as an off flag.
#
# tag_push_events comes back when releases ship. Change it here and in the
# app's handler together.
DESIRED = {
    "push_events": True,
    "merge_requests_events": True,
    "tag_push_events": False,
    "repository_update_events": False,
    "enable_ssl_verification": True,
}

EVENT_LABELS = [
    ("push_events", "push events"),
    ("merge_requests_events", "merge request events"),
    ("tag_push_events", "tag push events"),
    ("repository_update_events", "repository update"),
]

# Shown in Admin Area > System Hooks, so a colleague who finds this hook can
# tell what it is instead of deleting it to see what breaks. Older GitLab
# versions reject these two fields; see _write_hook for the fallback.
HOOK_NAME = "Jira development information"
HOOK_DESCRIPTION = (
    "Created by the Jira app's setup script. Sends push and merge request "
    "events to Jira. Delete this hook to disconnect."
)

ACTION_CREATE = "create"
ACTION_UPDATE = "update"
ACTION_REFUSE = "refuse"


class Refused(Exception):
    """A precondition failed. Nothing was written."""


class ApiError(Exception):
    """GitLab said no, or could not be reached."""


class TlsError(Exception):
    """The TLS certificate could not be verified."""


# --------------------------------------------------------------------------
# Pure helpers. No network, no credentials held, unit-tested in selftest.py.
# --------------------------------------------------------------------------

def fingerprint(value):
    """Short, non-reversible fingerprint of a secret, for display only.

    Same convention as the app's src/redact.js: 4 hex chars is 16 bits,
    enough to tell "this is the same secret Jira shows me" apart from "a
    different one", useless for an attack on a 256-bit secret.
    """
    if not value:
        return "(none)"
    return "sha256:" + hashlib.sha256(value.encode("utf-8")).hexdigest()[:4]


def scrub(text, secrets):
    """Replaces any secret that appears in text with its fingerprint.

    Applied to everything printed on a failure path. GitLab's own error
    bodies do not contain our credentials, and a Python traceback does not
    print local variables -- this is the belt to that pair of braces, so
    that "the token is never printed" holds even for a failure nobody
    anticipated.
    """
    out = text
    for secret in secrets:
        if secret:
            out = out.replace(secret, "[REDACTED %s]" % fingerprint(secret))
    return out


def normalize_hook_url(url):
    """Canonical form used to decide whether a hook is ours.

    Scheme and host are case-insensitive per RFC 3986; the path is not, and
    a Forge web trigger path segment is case-sensitive, so it is compared
    byte for byte. One trailing slash is ignored because the GitLab UI adds
    one on occasion. The fragment is dropped -- it is never sent.
    """
    parts = urlsplit(url.strip())
    path = parts.path
    if len(path) > 1 and path.endswith("/"):
        path = path[:-1]
    return urlunsplit((parts.scheme.lower(), parts.netloc.lower(), path,
                       parts.query, ""))


def host_of(url):
    return urlsplit(url).netloc.lower()


def decide(hooks, webtrigger_url):
    """Works out what to do, from the hooks the instance already has.

    Matching is on the web trigger URL and nothing else: GitLab never
    returns a hook's token (only token_present), so comparing secrets is
    impossible by construction.
    """
    target = normalize_hook_url(webtrigger_url)
    target_host = host_of(target)

    matches = []
    same_host_others = []
    for hook in hooks:
        raw = hook.get("url") or ""
        normalized = normalize_hook_url(raw)
        if normalized == target:
            matches.append(hook)
        elif host_of(normalized) == target_host:
            # A Forge web trigger URL's host identifies the app and
            # environment, so a different path on the same host is very
            # likely this same app on an older, rotated URL. "Very likely"
            # is enough to warn about and nowhere near enough to act on:
            # this only ever produces a line of output.
            same_host_others.append(hook)

    plan = {
        "webtrigger_url": webtrigger_url.strip(),
        "desired": dict(DESIRED),
        "same_host_others": same_host_others,
        "duplicates": [],
        "hook_id": None,
        "current": None,
    }

    if len(matches) == 0:
        plan["action"] = ACTION_CREATE
    elif len(matches) == 1:
        plan["action"] = ACTION_UPDATE
        plan["hook_id"] = matches[0].get("id")
        plan["current"] = matches[0]
    else:
        # Never guess which duplicate to keep, and never delete: this script
        # holds instance-wide admin rights for one run, and the extra hook
        # may not even be ours. Refusing costs the admin one click; guessing
        # wrong costs them either double deliveries or a deleted hook.
        plan["action"] = ACTION_REFUSE
        plan["duplicates"] = [h.get("id") for h in matches]

    return plan


def _flag(value):
    if value is True:
        return "on"
    if value is False:
        return "off"
    return "?"


def render_plan(plan, secret):
    """The dry-run output, and the same block printed before a real write."""
    lines = []
    action = plan["action"]

    if action == ACTION_REFUSE:
        ids = ", ".join("#%s" % i for i in plan["duplicates"])
        lines.append("Refusing to continue: %d system hooks already point at "
                     "this URL (%s)." % (len(plan["duplicates"]), ids))
        lines.append("Remove the duplicate in Admin Area > System Hooks, then "
                     "re-run.")
        lines.append("No changes were made.")
        return "\n".join(lines)

    if action == ACTION_UPDATE:
        lines.append("Would UPDATE system hook #%s  - matched on hook URL"
                     % plan["hook_id"])
    else:
        lines.append("Would CREATE a new system hook")
    lines.append("")

    lines.append("  Hook URL      %s" % plan["webtrigger_url"])
    lines.append("  Secret token  %s   (re-sent on every run)"
                 % fingerprint(secret))

    current = plan["current"] or {}
    first = True
    for key, label in EVENT_LABELS:
        prefix = "  Events      " if first else "              "
        first = False
        if action == ACTION_UPDATE:
            lines.append("%s  %-22s %-4s ->  %s"
                         % (prefix, label, _flag(current.get(key)),
                            _flag(plan["desired"][key])))
        else:
            lines.append("%s  %-22s %s"
                         % (prefix, label, _flag(plan["desired"][key])))

    if action == ACTION_UPDATE:
        lines.append("  SSL verify    %-22s %-4s ->  %s"
                     % ("", _flag(current.get("enable_ssl_verification")),
                        _flag(plan["desired"]["enable_ssl_verification"])))
    else:
        lines.append("  SSL verify    %-22s %s"
                     % ("", _flag(plan["desired"]["enable_ssl_verification"])))

    for hook in plan["same_host_others"]:
        lines.append("")
        lines.append("  Note: system hook #%s points at the same host with a "
                     "different path:" % hook.get("id"))
        lines.append("        %s" % hook.get("url"))
        lines.append("        That looks like this app on an older URL. It "
                     "will keep delivering.")
        lines.append("        This script will not touch it - remove it "
                     "yourself if it is stale.")

    return "\n".join(lines)


def check_gitlab_url(url):
    """Rejects a target we would not send an admin token to."""
    parts = urlsplit(url)
    if parts.scheme not in ("http", "https") or not parts.netloc:
        raise Refused("--gitlab-url must be a full URL, for example "
                      "https://gitlab.example.com")
    host = parts.hostname or ""
    if parts.scheme == "http" and host not in ("localhost", "127.0.0.1", "::1"):
        raise Refused(
            "Refusing to send an administrator token over plain http to %s.\n"
            "Use the https URL of your instance." % host)
    return url.rstrip("/")


def check_webtrigger_url(url):
    parts = urlsplit(url)
    if parts.scheme != "https" or not parts.netloc or len(parts.path) < 2:
        raise Refused("That does not look like a web trigger URL. Copy it "
                      "from the app's configuration screen in Jira; it "
                      "starts with https:// and ends in a long path segment.")
    return url.strip()


# --------------------------------------------------------------------------
# Transport. The only code that touches the network.
# --------------------------------------------------------------------------

class _NoRedirect(urllib.request.HTTPRedirectHandler):
    """Refuses every redirect.

    urllib re-sends request headers to the redirect target, so following a
    redirect would hand the PRIVATE-TOKEN header to whatever host the
    response names. An admin token is not something to replay at a location
    chosen by someone else, so a redirect is an error with an instruction
    rather than something handled transparently.
    """

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise ApiError(
            "%s redirected to %s.\nUse that URL directly with --gitlab-url; "
            "this script does not follow redirects, because that would send "
            "your token to a host you did not name." % (req.full_url, newurl))


def make_transport(ca_bundle=None):
    """Builds the callable Session uses: (method, url, headers, body) -> tuple.

    Replaced wholesale in selftest.py, which is why the credential handling
    can be tested without a GitLab.
    """
    context = ssl.create_default_context(cafile=ca_bundle)
    opener = urllib.request.build_opener(
        _NoRedirect(), urllib.request.HTTPSHandler(context=context))

    def transport(method, url, headers, body):
        request = urllib.request.Request(url, data=body, headers=headers,
                                         method=method)
        try:
            response = opener.open(request, timeout=30)
        except urllib.error.HTTPError as error:
            # A 4xx is an answer, not a transport failure: read the body so
            # GitLab's own message can be shown.
            return error.code, error.read()
        except urllib.error.URLError as error:
            reason = error.reason
            if isinstance(reason, ssl.SSLError):
                raise TlsError(
                    "Could not verify the TLS certificate of %s (%s).\n"
                    "If your instance uses an internal certificate authority, "
                    "pass its certificate with --ca-bundle /path/to/ca.pem."
                    % (host_of(url), reason))
            raise ApiError("Could not reach %s: %s" % (host_of(url), reason))
        with response:
            return response.getcode(), response.read()

    return transport


class Session:
    """Authenticated GitLab API calls. Holds the admin token; nothing else does.

    read_only=True makes any non-GET raise. That is what --dry-run runs on,
    so "a dry run performs no writes" is enforced here rather than merely
    being true of the current control flow.
    """

    def __init__(self, base_url, token, transport, read_only=False):
        self._base = base_url.rstrip("/")
        self._token = token
        self._transport = transport
        self._read_only = read_only

    def __repr__(self):
        # Explicit, so that no accident of formatting can put the token in a
        # string. The default repr would not have, but this is cheap.
        return "<Session %s read_only=%s>" % (self._base, self._read_only)

    def _call(self, method, path, payload=None):
        if self._read_only and method != "GET":
            raise RuntimeError(
                "read-only session refused a %s request" % method)

        url = self._base + "/api/v4" + path
        headers = {
            "PRIVATE-TOKEN": self._token,  # the token's only destination
            "Accept": "application/json",
            "User-Agent": "jira-gitlab-setup",
        }
        body = None
        if payload is not None:
            body = json.dumps(payload).encode("utf-8")
            headers["Content-Type"] = "application/json"

        status, raw = self._transport(method, url, headers, body)
        try:
            parsed = json.loads(raw.decode("utf-8")) if raw else None
        except ValueError:
            parsed = None

        if status >= 400:
            raise ApiError(_api_message(status, parsed, path))
        return parsed

    def get(self, path):
        return self._call("GET", path)

    def get_all(self, path):
        """Reads every page.

        Paging matters here: the default page size is 20, and a hook that
        exists on page 2 but is not looked for would be silently created a
        second time. Double delivery of every event is exactly the failure
        this script must not produce.
        """
        results = []
        page = 1
        while page <= 50:
            separator = "&" if "?" in path else "?"
            batch = self._call(
                "GET", "%s%sper_page=100&page=%d" % (path, separator, page))
            if not batch:
                break
            results.extend(batch)
            if len(batch) < 100:
                break
            page += 1
        return results

    def post(self, path, payload):
        return self._call("POST", path, payload)

    def put(self, path, payload):
        return self._call("PUT", path, payload)


def _api_message(status, parsed, path):
    detail = ""
    if isinstance(parsed, dict):
        detail = parsed.get("message") or parsed.get("error") or ""
        if not isinstance(detail, str):
            detail = json.dumps(detail)
    if status == 401:
        return ("GitLab rejected the token (401). It is invalid, expired, or "
                "lacks the 'api' scope.")
    if status == 403:
        return ("GitLab refused access (403) on %s. System hooks require an "
                "administrator token; this one is not an administrator's."
                % path)
    if status == 404 and path.startswith("/hooks/"):
        # Two causes, one remedy. Editing a system hook through the API
        # arrived in a later GitLab version than creating one, and
        # self-managed instances lag; the hook may also simply have been
        # deleted since the listing a moment ago. Recreating it fixes both,
        # and deleting-to-work-around is left to the admin rather than done
        # here on a guess.
        return ("PUT %s returned 404: either your GitLab is too old to edit "
                "a system hook through the API, or that hook no longer "
                "exists.\nDelete it in Admin Area > System Hooks if it is "
                "still there, then run this script again to recreate it."
                % path)
    return "GitLab returned %d on %s%s" % (
        status, path, (": " + detail) if detail else "")


# --------------------------------------------------------------------------
# Input. Kept at module level so selftest.py can replace them.
# --------------------------------------------------------------------------

def prompt_visible(text):
    """Reads a visible line, from the terminal even when stdin is a pipe."""
    if sys.stdin.isatty():
        return input(text).strip()
    try:
        tty = open("/dev/tty", "r+")
    except (IOError, OSError):
        raise Refused("No terminal available to ask for the web trigger URL. "
                      "Run this script interactively.")
    try:
        tty.write(text)
        tty.flush()
        return tty.readline().strip()
    finally:
        tty.close()


def prompt_secret(text):
    """No-echo read. getpass uses /dev/tty, so a piped stdin is fine."""
    return getpass.getpass(text).strip()


def read_token_from_stdin():
    line = sys.stdin.readline()
    if not line.strip():
        raise Refused("--token-stdin was given but stdin was empty.")
    return line.strip()


# The three values, in the order --stdin expects them.
STDIN_FIELDS = ("web trigger URL", "shared secret", "administrator token")


def read_all_from_stdin():
    """Reads the three inputs from stdin, one per line, for an unattended run.

    Why this exists as well as the prompts: with prompts only, the script
    cannot run where there is no terminal -- CI, configuration management,
    or a demonstration -- and something that cannot be exercised
    unattended does not get exercised.

    It keeps the same property the prompts have and the reason there is no
    --token flag: the values arrive on a file descriptor, not in the
    argument list, so they stay out of `ps` and out of shell history.
    Feed it from a file or a password manager, never from `echo`, which
    puts the secret in your history itself.
    """
    values = []
    for index, field in enumerate(STDIN_FIELDS):
        line = sys.stdin.readline()
        if not line.strip():
            raise Refused(
                "--stdin expects %d lines: %s.\nLine %d (%s) was empty or "
                "missing." % (len(STDIN_FIELDS), ", ".join(STDIN_FIELDS),
                              index + 1, field))
        values.append(line.strip())
    return values


# --------------------------------------------------------------------------
# The run itself.
# --------------------------------------------------------------------------

def parse_args(argv):
    parser = argparse.ArgumentParser(
        description="Register the Jira integration's GitLab system hook.",
        epilog="The administrator token is used for this run only and is "
               "never stored, printed or written to disk.")
    parser.add_argument("--gitlab-url", required=True,
                        help="Base URL of your GitLab, e.g. "
                             "https://gitlab.example.com")
    parser.add_argument("--dry-run", action="store_true",
                        help="Show exactly what would change and make no "
                             "write calls at all.")
    parser.add_argument("--token-stdin", action="store_true",
                        help="Read the administrator token from stdin instead "
                             "of prompting. The other two are still asked "
                             "for at the terminal.")
    parser.add_argument("--stdin", action="store_true",
                        help="Unattended: read all three inputs from stdin, "
                             "one per line, in the order %s."
                             % ", ".join(STDIN_FIELDS))
    parser.add_argument("--ca-bundle", metavar="PATH",
                        help="Certificate authority bundle, for an instance "
                             "with an internal certificate.")
    args = parser.parse_args(argv)
    if args.stdin and args.token_stdin:
        parser.error("--stdin already reads the token; drop --token-stdin.")
    return args


def _write_hook(session, plan, secret):
    """Creates or updates, with a fallback for older GitLab versions.

    name and description arrived in later GitLab versions. They are worth
    sending -- an unlabelled hook in the admin UI is one a colleague deletes
    to find out what it was -- but not worth failing over, so a 400 is
    retried once without them.
    """
    payload = dict(plan["desired"])
    payload["url"] = plan["webtrigger_url"]
    payload["token"] = secret

    labelled = dict(payload)
    labelled["name"] = HOOK_NAME
    labelled["description"] = HOOK_DESCRIPTION

    if plan["action"] == ACTION_CREATE:
        send = lambda body: session.post("/hooks", body)
    else:
        send = lambda body: session.put("/hooks/%s" % plan["hook_id"], body)

    try:
        return send(labelled)
    except ApiError as error:
        if "returned 400" not in str(error):
            raise
        return send(payload)


def _run(args, creds):
    gitlab_url = check_gitlab_url(args.gitlab_url)

    if args.stdin:
        webtrigger_raw, secret, token = read_all_from_stdin()
        creds["secret"] = secret
        creds["token"] = token
    else:
        print("Paste these two from the app's configuration screen in Jira.")
        webtrigger_raw = prompt_visible("  Web trigger URL: ")
        secret = prompt_secret("  Shared secret (not shown): ")
        creds["secret"] = secret
        if not secret:
            raise Refused("The shared secret is required. It is on the same "
                          "configuration screen as the web trigger URL.")
        if args.token_stdin:
            token = read_token_from_stdin()
        else:
            token = prompt_secret("  GitLab administrator token (not shown): ")
        creds["token"] = token
        if not token:
            raise Refused("An administrator token is required. System hooks "
                          "cannot be registered without one.")

    webtrigger_url = check_webtrigger_url(webtrigger_raw)

    session = Session(gitlab_url, token, make_transport(args.ca_bundle),
                      read_only=args.dry_run)

    version = session.get("/version") or {}
    user = session.get("/user") or {}
    # is_admin is only present for administrators on most versions, so its
    # absence is reported as unknown rather than as "no" -- the GET /hooks
    # below is the real test, and its 403 says so plainly.
    if user.get("is_admin") is True:
        admin = "yes"
    elif user.get("is_admin") is False:
        admin = "no"
    else:
        admin = "unknown"

    print("")
    print("Target instance : %s   (GitLab %s)"
          % (gitlab_url, version.get("version", "unknown")))
    print("Authenticated as: %s   (administrator: %s)"
          % (user.get("username", "unknown"), admin))
    print("")

    hooks = session.get_all("/hooks")
    plan = decide(hooks, webtrigger_url)
    print(render_plan(plan, secret))
    print("")

    if plan["action"] == ACTION_REFUSE:
        return EXIT_REFUSED

    if args.dry_run:
        print("DRY RUN - no changes were made. Re-run without --dry-run to "
              "apply.")
        return EXIT_OK

    result = _write_hook(session, plan, secret) or {}
    hook_id = result.get("id", plan["hook_id"])
    verb = "created" if plan["action"] == ACTION_CREATE else "updated"
    print("Applied. System hook #%s %s. Push and merge request events from "
          "every project on this instance now go to Jira." % (hook_id, verb))
    print("")
    print("The administrator token was used for this run only. It was not "
          "written to disk, never appeared in the command line, and nothing "
          "of it remains once this process exits.")
    print("If you created it for this setup, you can revoke it now:")
    print("  %s/-/user_settings/personal_access_tokens" % gitlab_url)
    return EXIT_OK


def main(argv=None):
    args = parse_args(argv)

    # Holds the two secrets as soon as they are read, so that the failure
    # handlers below can scrub them out of anything they print. Before they
    # are read there is nothing to scrub; after the process exits there is
    # nothing left at all.
    creds = {"token": None, "secret": None}

    try:
        return _run(args, creds)
    except Refused as error:
        print(str(error), file=sys.stderr)
        print("No changes were made.", file=sys.stderr)
        return EXIT_REFUSED
    except TlsError as error:
        print(scrub(str(error), [creds["token"], creds["secret"]]),
              file=sys.stderr)
        print("No changes were made.", file=sys.stderr)
        return EXIT_TLS
    except ApiError as error:
        print(scrub(str(error), [creds["token"], creds["secret"]]),
              file=sys.stderr)
        print("No changes were made.", file=sys.stderr)
        return EXIT_API
    except KeyboardInterrupt:
        print("\nInterrupted. No changes were made.", file=sys.stderr)
        return EXIT_REFUSED
    except Exception:  # noqa: BLE001 - deliberate, see scrub()
        print(scrub(traceback.format_exc(),
                    [creds["token"], creds["secret"]]), file=sys.stderr)
        print("Unexpected error. No changes were made.", file=sys.stderr)
        return EXIT_API


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