riposte-marketplace/integrations/microsoft-graph-files/scripts/update_site_permissions.py

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

S = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
I = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))

GRAPH = (S.get("host") or "https://graph.microsoft.com").rstrip("/")
LOGIN = (S.get("login_url") or "https://login.microsoftonline.com").rstrip("/")


def out(value):
    print(json.dumps(value, default=str))


def fail(message, **extra):
    payload = {"error": message}
    payload.update(extra)
    print(json.dumps(payload, default=str))
    sys.exit(1)


def _bool(v):
    return str(v).lower() in ("1", "true", "yes")


def _list(v):
    if isinstance(v, list):
        return v
    return [x.strip() for x in str(v).split(",") if x.strip()]


def _ctx():
    return ssl._create_unverified_context() if _bool(S.get("insecure")) else ssl.create_default_context()


_TOKEN = None


def token():
    global _TOKEN
    if _TOKEN is not None:
        return _TOKEN
    data = urllib.parse.urlencode({
        "client_id": S.get("client_id", ""),
        "client_secret": S.get("client_secret", ""),
        "grant_type": "client_credentials",
        "scope": GRAPH + "/.default",
    }).encode("utf-8")
    url = LOGIN + "/" + urllib.parse.quote(S.get("tenant_id", "")) + "/oauth2/v2.0/token"
    req = urllib.request.Request(url, data=data,
                                 headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=60, context=_ctx()) as r:
            _TOKEN = json.loads(r.read())["access_token"]
    except urllib.error.HTTPError as e:
        raise Exception("Authentication failed: HTTP " + str(e.code) + " " + e.read().decode("utf-8", "replace"))
    return _TOKEN


def graph(method, path, params=None, json_body=None, data=None, headers=None, resp="json"):
    url = path if path.startswith("http") else (GRAPH + "/v1.0" + path)
    if params:
        clean = {k: v for k, v in params.items() if v not in (None, "")}
        if clean:
            url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean)
    h = {"Authorization": "Bearer " + token(), "Accept": "application/json"}
    body = None
    if json_body is not None:
        body = json.dumps(json_body).encode("utf-8")
        h["Content-Type"] = "application/json"
    elif data is not None:
        body = data
        h["Content-Type"] = "application/octet-stream"
    if headers:
        h.update(headers)
    req = urllib.request.Request(url, data=body, headers=h, method=method)
    try:
        with urllib.request.urlopen(req, timeout=180, context=_ctx()) as r:
            raw = r.read()
            if resp == "response":
                return {"status": r.status}
            if resp == "content":
                return raw
            return json.loads(raw) if raw else {}
    except urllib.error.HTTPError as e:
        raise Exception("Graph HTTP " + str(e.code) + ": " + e.read().decode("utf-8", "replace"))


def fetch_url(url):
    """Fetch raw bytes from an arbitrary URL (no Graph auth header)."""
    req = urllib.request.Request(url, method="GET")
    with urllib.request.urlopen(req, timeout=180, context=_ctx()) as r:
        return r.read()


def input_bytes():
    """Resolve upload content from content_base64 or source_url."""
    if I.get("content_base64"):
        return base64.b64decode(I["content_base64"])
    if I.get("source_url"):
        return fetch_url(I["source_url"])
    fail("Provide either content_base64 or source_url")


def col_letter(n):
    s = ""
    n = int(n)
    while n > 0:
        n, rem = divmod(n - 1, 26)
        s = chr(65 + rem) + s
    return s


def resolve_site_id():
    if I.get("site_id"):
        return I["site_id"]
    name = I.get("site_name")
    if not name:
        fail("Provide site_id or site_name")
    res = graph("GET", "/sites", params={"search": name})
    value = res.get("value", [])
    if not value:
        fail("Site not found: " + name)
    return value[0]["id"]


def resolve_list_id(site_id):
    if I.get("list_id"):
        return I["list_id"]
    name = I.get("list_name")
    if not name:
        fail("Provide list_id or list_name")
    res = graph("GET", "/sites/" + site_id + "/lists", params={"$top": 999})
    for lst in res.get("value", []):
        if (lst.get("displayName") or lst.get("name") or "").lower() == name.lower():
            return lst["id"]
    fail("List not found: " + name)


def drive_item_base(object_type, object_type_id, item_id):
    if object_type == "drives":
        return "/drives/" + object_type_id + "/items/" + item_id
    return "/" + object_type + "/" + object_type_id + "/drive/items/" + item_id


def worksheet_base(object_type, object_type_id, item_id, worksheet):
    ws = urllib.parse.quote(worksheet)
    if object_type == "drives":
        return "/drives/" + object_type_id + "/items/" + item_id + "/workbook/worksheets/" + ws
    return "/" + object_type + "/" + object_type_id + "/drive/items/" + item_id + "/workbook/worksheets/" + ws



def run():
    site_id = resolve_site_id()
    uri = "/sites/" + site_id + "/permissions/" + I["permission_id"]
    out(graph("PATCH", uri, json_body={"roles": _list(I["role"])}))


try:
    run()
except SystemExit:
    raise
except Exception as e:
    fail(str(e))