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 # The call queue lives under ONE OF TWO PRODUCTS, and which one is not something an # operator can see: the control panel says "File d'appels" for both. OVH files them # under easyHunting or under ovhPabx depending on the offer the number carries, and # the hunting routes underneath are identical either way. # # Assuming easyHunting is what this did first, and the symptom was a 404 reading # "This service does not exist" — which sounds like a wrong number and is in fact a # wrong product. So it is DISCOVERED, once per process, and a failure says what does # exist rather than what does not. _PRODUCTS = ("easyHunting", "ovhPabx") def _product(cfg): if _CACHE.get("product"): return _CACHE["product"] ba, service = _telephony(cfg) found = {} for product in _PRODUCTS: names = request("GET", "/telephony/" + urllib.parse.quote(ba, safe="") + "/" + product, cfg) or [] found[product] = [str(n) for n in names] if service in found[product]: _CACHE["product"] = product return product raise Exception( "no call queue named %s on %s. easyHunting holds [%s]; ovhPabx holds [%s]. " "Set telephony_service to one of those names, or configure the number as a " "call queue in the control panel first." % (service, ba, ", ".join(found.get("easyHunting") or ["-"]), ", ".join(found.get("ovhPabx") or ["-"]))) def _base(cfg): ba, service = _telephony(cfg) return ("/telephony/" + urllib.parse.quote(ba, safe="") + "/" + _product(cfg) + "/" + 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]) def _ovh_number(raw): """OVH writes external numbers in 00-prefixed international form. A "+" is accepted nowhere in a phoneNumber field, so a ladder handed straight from Riposte - which speaks E.164 - would be refused number by number, and the on-call ladder would silently stay as it was. """ s = "".join(ch for ch in str(raw) if ch.isdigit() or ch == "+").strip() if s.startswith("+"): return "00" + s[1:] return s def _e164(raw): """The inverse, so a readback can be compared with what Riposte published.""" s = "".join(ch for ch in str(raw) if ch.isdigit() or ch == "+").strip() if s.startswith("+"): return s if s.startswith("00"): return "+" + s[2:] return s def _agents(cfg): """Every agent of the service, as {agentId: agent}.""" out = {} for agent_id in request("GET", _base(cfg) + "/hunting/agent", cfg) or []: out[str(agent_id)] = request("GET", _base(cfg) + "/hunting/agent/" + str(agent_id), cfg) return out # ── list_calls ────────────────────────────────────────────────────────────── # # The calls the carrier answered on the on-call number, so the pager's journal can # be filled after the fact. # # AFTER THE FACT IS THE POINT AND THE LIMIT. In this mode the carrier answers, so # Riposte learns nothing while the call is happening: the journal is a # rapatriation, minutes to hours late, and it cannot say which rung took the call — # a consumption record knows the number dialled and how long it lasted, not which # agent the queue reached. Honest and useful; not the live chain the Asterisk mode # writes. def main(cfg, inputs): ba, service = _telephony(cfg) limit = int(inputs.get("limit") or 50) limit = max(1, min(limit, 200)) since = str(inputs.get("since") or "").strip() if not since: # A day, which is the window somebody actually asks about ("nobody called # back last night"). import datetime since = (datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%S+00:00") path = ("/telephony/" + urllib.parse.quote(ba, safe="") + "/service/" + urllib.parse.quote(service, safe="") + "/voiceConsumption") query = urllib.parse.urlencode({"creationDatetime.from": since, "wayType": "incoming"}) ids = request("GET", path + "?" + query, cfg) or [] # BOUNDED, and the truncation is REPORTED. A silent top-N on a journal reads as # "this is everything that happened", which is the one sentence a journal must # never say by accident. truncated = len(ids) > limit calls = [] for consumption_id in ids[:limit]: c = request("GET", path + "/" + str(consumption_id), cfg) calls.append({ "id": str(c.get("consumptionId") or consumption_id), "from_number": _e164(c.get("calling")), "did": _e164(c.get("called") or c.get("dialed")), "started_at": c.get("creationDatetime"), "duration": c.get("duration"), # The carrier's own word for how it ended. Kept verbatim rather than # translated into ours: it is what their support will ask about. "hangup_nature": c.get("hangupNature"), }) return {"ok": True, "calls": calls, "count": len(calls), "total": len(ids), "truncated": truncated, "since": since} _run(main)