summary refs log tree commit diff
path: root/scripts/federation_client.py
diff options
context:
space:
mode:
Diffstat (limited to 'scripts/federation_client.py')
-rw-r--r--scripts/federation_client.py146
1 files changed, 0 insertions, 146 deletions
diff --git a/scripts/federation_client.py b/scripts/federation_client.py
deleted file mode 100644
index ea62dceb36..0000000000
--- a/scripts/federation_client.py
+++ /dev/null
@@ -1,146 +0,0 @@
-import nacl.signing
-import json
-import base64
-import requests
-import sys
-import srvlookup
-
-
-def encode_base64(input_bytes):
-    """Encode bytes as a base64 string without any padding."""
-
-    input_len = len(input_bytes)
-    output_len = 4 * ((input_len + 2) // 3) + (input_len + 2) % 3 - 2
-    output_bytes = base64.b64encode(input_bytes)
-    output_string = output_bytes[:output_len].decode("ascii")
-    return output_string
-
-
-def decode_base64(input_string):
-    """Decode a base64 string to bytes inferring padding from the length of the
-    string."""
-
-    input_bytes = input_string.encode("ascii")
-    input_len = len(input_bytes)
-    padding = b"=" * (3 - ((input_len + 3) % 4))
-    output_len = 3 * ((input_len + 2) // 4) + (input_len + 2) % 4 - 2
-    output_bytes = base64.b64decode(input_bytes + padding)
-    return output_bytes[:output_len]
-
-
-def encode_canonical_json(value):
-    return json.dumps(
-         value,
-         # Encode code-points outside of ASCII as UTF-8 rather than \u escapes
-         ensure_ascii=False,
-         # Remove unecessary white space.
-         separators=(',',':'),
-         # Sort the keys of dictionaries.
-         sort_keys=True,
-         # Encode the resulting unicode as UTF-8 bytes.
-     ).encode("UTF-8")
-
-
-def sign_json(json_object, signing_key, signing_name):
-    signatures = json_object.pop("signatures", {})
-    unsigned = json_object.pop("unsigned", None)
-
-    signed = signing_key.sign(encode_canonical_json(json_object))
-    signature_base64 = encode_base64(signed.signature)
-
-    key_id = "%s:%s" % (signing_key.alg, signing_key.version)
-    signatures.setdefault(signing_name, {})[key_id] = signature_base64
-
-    json_object["signatures"] = signatures
-    if unsigned is not None:
-        json_object["unsigned"] = unsigned
-
-    return json_object
-
-
-NACL_ED25519 = "ed25519"
-
-def decode_signing_key_base64(algorithm, version, key_base64):
-    """Decode a base64 encoded signing key
-    Args:
-        algorithm (str): The algorithm the key is for (currently "ed25519").
-        version (str): Identifies this key out of the keys for this entity.
-        key_base64 (str): Base64 encoded bytes of the key.
-    Returns:
-        A SigningKey object.
-    """
-    if algorithm == NACL_ED25519:
-        key_bytes = decode_base64(key_base64)
-        key = nacl.signing.SigningKey(key_bytes)
-        key.version = version
-        key.alg = NACL_ED25519
-        return key
-    else:
-        raise ValueError("Unsupported algorithm %s" % (algorithm,))
-
-
-def read_signing_keys(stream):
-    """Reads a list of keys from a stream
-    Args:
-        stream : A stream to iterate for keys.
-    Returns:
-        list of SigningKey objects.
-    """
-    keys = []
-    for line in stream:
-        algorithm, version, key_base64 = line.split()
-        keys.append(decode_signing_key_base64(algorithm, version, key_base64))
-    return keys
-
-
-def lookup(destination, path):
-    if ":" in destination:
-        return "https://%s%s" % (destination, path)
-    else:
-        try:
-            srv = srvlookup.lookup("matrix", "tcp", destination)[0]
-            return "https://%s:%d%s" % (srv.host, srv.port, path)
-        except:
-            return "https://%s:%d%s" % (destination, 8448, path)
-
-def get_json(origin_name, origin_key, destination, path):
-    request_json = {
-        "method": "GET",
-        "uri": path,
-        "origin": origin_name,
-        "destination": destination,
-    }
-
-    signed_json = sign_json(request_json, origin_key, origin_name)
-
-    authorization_headers = []
-
-    for key, sig in signed_json["signatures"][origin_name].items():
-        authorization_headers.append(bytes(
-            "X-Matrix origin=%s,key=\"%s\",sig=\"%s\"" % (
-                origin_name, key, sig,
-            )
-        ))
-
-    result = requests.get(
-        lookup(destination, path),
-        headers={"Authorization": authorization_headers[0]},
-        verify=False,
-    )
-    return result.json()
-
-
-def main():
-    origin_name, keyfile, destination, path = sys.argv[1:]
-
-    with open(keyfile) as f:
-        key = read_signing_keys(f)[0]
-
-    result = get_json(
-        origin_name, key, destination, "/_matrix/federation/v1/" + path
-    )
-
-    json.dump(result, sys.stdout)
-
-if __name__ == "__main__":
-    main()