123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246 |
- from __future__ import print_function
- import argparse
- import nacl.signing
- import json
- import base64
- import requests
- import sys
- import srvlookup
- import yaml
- 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,
-
- ensure_ascii=False,
-
- separators=(',',':'),
-
- sort_keys=True,
-
- ).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 request_json(method, origin_name, origin_key, destination, path, content):
- if method is None:
- if content is None:
- method = "GET"
- else:
- method = "POST"
- json_to_sign = {
- "method": method,
- "uri": path,
- "origin": origin_name,
- "destination": destination,
- }
- if content is not None:
- json_to_sign["content"] = json.loads(content)
- signed_json = sign_json(json_to_sign, origin_key, origin_name)
- authorization_headers = []
- for key, sig in signed_json["signatures"][origin_name].items():
- header = "X-Matrix origin=%s,key=\"%s\",sig=\"%s\"" % (
- origin_name, key, sig,
- )
- authorization_headers.append(bytes(header))
- print ("Authorization: %s" % header, file=sys.stderr)
- dest = lookup(destination, path)
- print ("Requesting %s" % dest, file=sys.stderr)
- result = requests.request(
- method=method,
- url=dest,
- headers={"Authorization": authorization_headers[0]},
- verify=False,
- data=content,
- )
- sys.stderr.write("Status Code: %d\n" % (result.status_code,))
- return result.json()
- def main():
- parser = argparse.ArgumentParser(
- description=
- "Signs and sends a federation request to a matrix homeserver",
- )
- parser.add_argument(
- "-N", "--server-name",
- help="Name to give as the local homeserver. If unspecified, will be "
- "read from the config file.",
- )
- parser.add_argument(
- "-k", "--signing-key-path",
- help="Path to the file containing the private ed25519 key to sign the "
- "request with.",
- )
- parser.add_argument(
- "-c", "--config",
- default="homeserver.yaml",
- help="Path to server config file. Ignored if --server-name and "
- "--signing-key-path are both given.",
- )
- parser.add_argument(
- "-d", "--destination",
- default="matrix.org",
- help="name of the remote homeserver. We will do SRV lookups and "
- "connect appropriately.",
- )
- parser.add_argument(
- "-X", "--method",
- help="HTTP method to use for the request. Defaults to GET if --data is"
- "unspecified, POST if it is."
- )
- parser.add_argument(
- "--body",
- help="Data to send as the body of the HTTP request"
- )
- parser.add_argument(
- "path",
- help="request path. We will add '/_matrix/federation/v1/' to this."
- )
- args = parser.parse_args()
- if not args.server_name or not args.signing_key_path:
- read_args_from_config(args)
- with open(args.signing_key_path) as f:
- key = read_signing_keys(f)[0]
- result = request_json(
- args.method,
- args.server_name, key, args.destination,
- "/_matrix/federation/v1/" + args.path,
- content=args.body,
- )
- json.dump(result, sys.stdout)
- print ("")
- def read_args_from_config(args):
- with open(args.config, 'r') as fh:
- config = yaml.safe_load(fh)
- if not args.server_name:
- args.server_name = config['server_name']
- if not args.signing_key_path:
- args.signing_key_path = config['signing_key_path']
- if __name__ == "__main__":
- main()
|