riposte-marketplace/integrations/jira/scripts/list_transitions.py

import base64, html, json, os, re, sys, urllib.parse, urllib.request, urllib.error


def _cfg():
    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))


def _is_cloud():
    return bool(str(_cfg().get("cloud_id") or "").strip())


def _api():
    return "3" if _is_cloud() else "2"


def _base():
    cfg = _cfg()
    root = str(cfg.get("server_url") or "https://api.atlassian.com/ex/jira").rstrip("/")
    cloud_id = str(cfg.get("cloud_id") or "").strip()
    if not cloud_id and "api.atlassian.com" in root:
        # The Cloud gateway addresses a site by its cloud id; without one every
        # path below it is a dead link, and Jira answers with an HTML 404 that
        # says nothing about the cause.
        raise Exception(
            "server_url is the Jira Cloud gateway (" + root + ") but cloud_id is empty. "
            "Set cloud_id, found at admin.atlassian.com/s/<cloud_id>/users, or point "
            "server_url at your own Jira URL for On-Prem/Data Center."
        )
    if cloud_id and "api.atlassian.com" not in root:
        # The reverse mistake, and the harder one to see: against any other host
        # a cloud id is just a path segment, and every call 404s.
        raise Exception(
            "cloud_id is set but server_url (" + root + ") is not the Jira Cloud gateway. "
            "A cloud id addresses a site through https://api.atlassian.com/ex/jira only. "
            "Clear cloud_id for On-Prem/Data Center, or restore the gateway URL."
        )
    return root + "/" + cloud_id if cloud_id else root


def _creds():
    # A credential pasted from a password manager often carries a trailing
    # newline; unstripped it corrupts the header and reads back as a wrong
    # password. Basic wins whenever the pair is complete: a token left behind
    # from another deployment must not outrank credentials just filled in.
    # A lone username, or a lone api_key, is not a pair.
    cfg = _cfg()
    user = str(cfg.get("username") or "").strip()
    key = str(cfg.get("api_key") or "").strip()
    pat = str(cfg.get("pat") or "").strip()
    mode = "basic" if (user and key) else ("pat" if pat else "")
    return mode, user, key, pat


def _auth_method():
    # Named in a 401/403, where the useful question is which credential was sent.
    mode, user, key, pat = _creds()
    if mode == "basic":
        return "Basic authentication as " + user + (
            " (a personal access token is also configured and was ignored)" if pat else ""
        )
    if mode == "pat":
        half = ""
        if user or key:
            half = " (" + ("username" if user else "api_key") + " is set but its pair is "
            half += "empty, so Basic authentication was not used)"
        return "a personal access token" + half
    return "no credential"


def _auth():
    mode, user, key, pat = _creds()
    if mode == "basic":
        cred = (user + ":" + key).encode("utf-8")
        return "Basic " + base64.b64encode(cred).decode("ascii")
    if mode == "pat":
        return "Bearer " + pat
    raise Exception(
        "No usable credential. Jira Cloud needs username (account email) together with "
        "api_key (API token); On-Prem/Data Center needs pat. "
        + ("Only " + ("username" if user else "api_key") + " is set."
           if (user or key) else "None of them is set.")
    )


def _summarize(raw, ctype):
    # An error page is a wall of markup around one sentence. Keep the <title>
    # and the readable text: the rest only makes a log impossible to read.
    text = raw.decode("utf-8", "replace") if isinstance(raw, bytes) else str(raw)
    if "json" in (ctype or "").lower():
        return text.strip()[:500]
    title = re.search(r"<title[^>]*>(.*?)</title>", text, re.S | re.I)
    body = re.sub(r"<(script|style|head)[^>]*>.*?</\1>", " ", text, flags=re.S | re.I)
    body = re.sub(r"<[^>]+>", " ", body)
    out = (title.group(1) + " \u2014 " if title else "") + body
    return re.sub(r"\s+", " ", html.unescape(out)).strip()[:500]


def request(method, path, params=None, body=None, full_url=None, raw=False):
    url = full_url or (_base() + path)
    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
    if q:
        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
    data = json.dumps(body).encode("utf-8") if body is not None else None
    headers = {"Accept": "application/json", "Authorization": _auth()}
    if data is not None:
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    with urllib.request.urlopen(req, timeout=90) as r:
        content = r.read()
        if raw:
            return content
        if not content:
            return {}
        try:
            return json.loads(content)
        except ValueError:
            # A 2xx that is not JSON is a login page, an SSO redirect or a proxy
            # answering in place of Jira. The URL the redirects landed on is
            # usually the whole diagnosis.
            landed = urllib.parse.urlsplit(r.geturl())
            ctype = r.headers.get("Content-Type") or "unknown"
            raise Exception(
                "Jira answered " + str(r.getcode()) + " with a non-JSON body at "
                + landed.scheme + "://" + landed.netloc + landed.path
                + " (content-type: " + ctype + "): " + _summarize(content, ctype)
            )


def issue_ref(inputs):
    ref = str(inputs.get("issue_id") or "") or str(inputs.get("issue_key") or "")
    if not ref:
        raise Exception("issue_id or issue_key is required")
    return urllib.parse.quote(ref, safe="")


def adf(text):
    # Jira Cloud (API v3) expects Atlassian Document Format for rich text; On-Prem (v2) takes plain text.
    if not _is_cloud():
        return text
    return {"type": "doc", "version": 1,
            "content": [{"type": "paragraph", "content": [{"type": "text", "text": text}]}]}


def main():
    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
    ref = issue_ref(inputs)
    raw_ref = str(inputs.get("issue_id") or "") or str(inputs.get("issue_key") or "")

    res = request("GET", "/rest/api/" + _api() + "/issue/" + ref + "/transitions")
    transitions = [
        {"id": t.get("id"), "name": t.get("name"), "to": (t.get("to") or {}).get("name", "")}
        for t in (res.get("transitions") or [])
    ]
    print(json.dumps({"issue": raw_ref, "transitions": transitions}))


try:
    main()
except urllib.error.HTTPError as e:
    ctype = e.headers.get("Content-Type", "") if e.headers else ""
    detail = _summarize(e.read(), ctype)
    if e.code in (401, 403):
        detail += " \u2014 the request was sent using " + _auth_method()
    print(json.dumps({"error": "HTTP " + str(e.code), "detail": detail}))
    sys.exit(1)
except Exception as e:
    print(json.dumps({"error": str(e)}))
    sys.exit(1)