riposte-marketplace/integrations/ovhcloud/scripts/test_connection.py

import json, os, sys, hashlib, time, urllib.parse, urllib.request, urllib.error


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


def _inputs():
    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))


# ── authentication ──────────────────────────────────────────────────────────
#
# TWO SCHEMES, because OVHcloud now offers two and an operator does not always get
# to choose. The historical one signs each request with an application key/secret
# plus a consumer key. The newer SERVICE ACCOUNT is OAuth2 client-credentials, and
# it is what the control panel offers when application keys are not available on
# the account.
#
# Which one is in use is decided by what was filled in, not by a mode selector:
# a client_id and a client_secret mean the service account, anything else falls
# back to the signed scheme. One less field to get wrong, and an instance created
# before service accounts existed keeps working untouched.
_TOKEN_HOSTS = {
    "eu": "https://www.ovh.com",
    "ca": "https://ca.ovh.com",
    "us": "https://us.ovhcloud.com",
}
_CACHE = {}


def _oauth(cfg):
    return bool(str(cfg.get("client_id") or "").strip() and str(cfg.get("client_secret") or "").strip())


def _endpoint(cfg):
    """The API base, corrected to match the credentials in use.

    The same routes answer under /1.0 and under /v1; the first is the signed
    scheme's historical spelling and the second is the one OAuth2 documents. An
    operator should not have to know that, so a service account gets /v1 whatever
    was typed — a Bearer token sent to the legacy spelling is the kind of failure
    that reads as "bad credentials" and sends somebody to regenerate a perfectly
    good secret.
    """
    raw = (str(cfg.get("endpoint") or "https://eu.api.ovh.com/1.0")).rstrip("/")
    if not _oauth(cfg):
        return raw
    if raw.endswith("/1.0"):
        return raw[:-4] + "/v1"
    if raw.endswith("/v1") or raw.endswith("/v2"):
        return raw
    return raw + "/v1"


def _token_url(cfg):
    explicit = str(cfg.get("oauth_token_url") or "").strip()
    if explicit:
        return explicit
    host = urllib.parse.urlparse(_endpoint(cfg)).netloc
    region = host.split(".")[0] if host else "eu"
    return _TOKEN_HOSTS.get(region, _TOKEN_HOSTS["eu"]) + "/auth/oauth2/token"


def _bearer(cfg):
    """The service account's access token.

    CACHED FOR THE LIFE OF THE PROCESS, which is not a micro-optimisation: publishing
    an on-call ladder makes a dozen or more API calls, and a token request in front
    of each one would triple the traffic and the time — on the path that runs every
    time the rota turns.
    """
    if _CACHE.get("token"):
        return _CACHE["token"]
    data = urllib.parse.urlencode({
        "grant_type": "client_credentials",
        "client_id": str(cfg.get("client_id", "")).strip(),
        "client_secret": str(cfg.get("client_secret", "")).strip(),
        # The scope the service account was granted in the control panel is what
        # actually bounds it; "all" here asks for everything that account may do.
        # Narrow the SERVICE ACCOUNT's own IAM policy rather than this value.
        "scope": "all",
    }).encode("utf-8")
    req = urllib.request.Request(
        _token_url(cfg), data=data, method="POST",
        headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=30) as r:
        payload = json.loads(r.read() or b"{}")
    token = payload.get("access_token")
    if not token:
        raise Exception("the service account returned no access token: " + json.dumps(payload)[:200])
    _CACHE["token"] = token
    return token


def _time_offset(cfg):
    """Seconds between this clock and OVH's, read once.

    The signature carries a timestamp and OVH refuses one that has drifted, so the
    server's own time is the safe source. Read ONCE and applied as an offset rather
    than fetched per request: the ladder makes many calls, and an extra round trip
    in front of each was doubling them.
    """
    if "offset" not in _CACHE:
        req = urllib.request.Request(_endpoint(cfg) + "/auth/time", method="GET")
        with urllib.request.urlopen(req, timeout=30) as r:
            server = int(r.read().decode("utf-8", "replace").strip())
        _CACHE["offset"] = server - int(time.time())
    return _CACHE["offset"]


def request(method, path, cfg, body=None):
    url = _endpoint(cfg) + path
    body_str = json.dumps(body) if body is not None else ""
    headers = {"Accept": "application/json"}
    if _oauth(cfg):
        headers["Authorization"] = "Bearer " + _bearer(cfg)
    else:
        app_key = str(cfg.get("application_key", "")).strip()
        app_secret = str(cfg.get("application_secret", "")).strip()
        consumer = str(cfg.get("consumer_key", "")).strip()
        if not (app_key and app_secret and consumer):
            raise Exception(
                "this instance has no credentials: fill in either a service account "
                "(client_id + client_secret) or the application key, secret and consumer key")
        ts = str(int(time.time()) + _time_offset(cfg))
        to_sign = app_secret + "+" + consumer + "+" + method.upper() + "+" + url + "+" + body_str + "+" + ts
        headers["X-Ovh-Application"] = app_key
        headers["X-Ovh-Consumer"] = consumer
        headers["X-Ovh-Timestamp"] = ts
        headers["X-Ovh-Signature"] = "$1$" + hashlib.sha1(to_sign.encode("utf-8")).hexdigest()
    data = None
    if body is not None:
        data = body_str.encode("utf-8")
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    with urllib.request.urlopen(req, timeout=60) as r:
        raw = r.read()
        return json.loads(raw) if raw else {}


def _run(fn):
    try:
        print(json.dumps(fn(_cfg(), _inputs())))
    except urllib.error.HTTPError as e:
        print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
        sys.exit(1)
    except Exception as e:
        print(json.dumps({"error": str(e)}))
        sys.exit(1)


# ── the call queue ──────────────────────────────────────────────────────────
#
# THE SAME COORDINATES FOR EVERY ON-CALL COMMAND, hence this block: the billing
# account, the service (the number the queue hangs from) and the queue itself.
# Repeated in each script rather than imported, because the marketplace collects
# loose *.py files and every command has to stand on its own.
def _telephony(cfg):
    ba = str(cfg.get("telephony_billing_account") or "").strip()
    service = str(cfg.get("telephony_service") or "").strip()
    if not ba or not service:
        raise Exception(
            "this instance has no telephony coordinates: set telephony_billing_account "
            "(e.g. fr-xxxxx-1) and telephony_service (the call queue's number, 0033xxxxxxxxx)")
    return ba, service


def _base(cfg):
    ba, service = _telephony(cfg)
    return "/telephony/" + urllib.parse.quote(ba, safe="") + "/easyHunting/" + urllib.parse.quote(service, safe="")


def _queue_id(cfg):
    """The queue to act on.

    Configured explicitly, or discovered when the service holds exactly one, which is
    the normal shape. AMBIGUITY IS REFUSED RATHER THAN GUESSED: picking the first of
    several would publish the on-call ladder into whichever queue OVH happened to
    list first, and the symptom is a number that rings the wrong team at 03:00.
    """
    configured = str(cfg.get("telephony_queue_id") or "").strip()
    if configured:
        return configured
    ids = request("GET", _base(cfg) + "/hunting/queue", cfg)
    if not isinstance(ids, list) or not ids:
        raise Exception("this service holds no call queue; create one in the control panel first")
    if len(ids) > 1:
        raise Exception("this service holds %d queues (%s) - set telephony_queue_id to say which one" %
                        (len(ids), ", ".join(str(i) for i in ids)))
    return str(ids[0])



# ── the probe ───────────────────────────────────────────────────────────────
#
# IT TESTS WHAT THIS INSTANCE IS FOR, which is not always the same thing.
#
# It used to call /me, unconditionally. That reads the ACCOUNT, and an instance
# created for on-call routing has no business reading the account — a service
# account narrowed to /telephony, which is exactly what it should be narrowed to,
# answers 403 "User not granted for this request" on account:apiovh:me/get. The
# credentials were perfect and the test said they were not, which is the worst
# thing a test button can do: it sends somebody off to regenerate a secret that was
# never wrong.
#
# So: telephony coordinates configured means the telephony probe. That one is worth
# more than /me ever was — it proves the credentials, the IAM policy, the billing
# account, the service AND the queue in a single call.
def main(cfg, inputs):
    ba = str(cfg.get("telephony_billing_account") or "").strip()
    service = str(cfg.get("telephony_service") or "").strip()
    scheme = "service account" if _oauth(cfg) else "application key"
    if ba and service:
        queues = request("GET", _base(cfg) + "/hunting/queue", cfg)
        if not isinstance(queues, list) or not queues:
            # Reached the API and found the service, but it holds no queue. NOT a
            # failure of the credentials, and saying so plainly is the difference
            # between "go and create a queue" and an evening spent on authentication.
            return {"ok": True, "auth": scheme, "telephony": True, "queues": 0,
                    "detail": "Authenticated, and the service was found, but it holds no call queue yet. "
                              "Create one in the control panel before publishing an on-call ladder."}
        return {"ok": True, "auth": scheme, "telephony": True,
                "queues": len(queues), "queue_ids": [str(q) for q in queues]}
    resp = request("GET", "/me", cfg)
    return {"ok": True, "auth": scheme, "nichandle": resp.get("nichandle")}


_run(main)