- using resolver

- testing move to json configs

Author: bergm

zalfmas_common/common.py

@@ -98,37 +98,55 @@ def update_config(config, argv, print_config=False, allow_new_keys=False):
             print(config)
 
 
-def create_service_toml_config(header_comment: str = None, id = None, name=None, description=None,
-                               host=None, port=None, serve_bootstrap=None, sturdy_ref_token=None,
-                               reg_sturdy_ref=None, reg_category=None, **kwargs) -> tk.TOMLDocument:
-    config = tk.document()
-    config.add(tk.comment(header_comment if header_comment else "MAS service configuration file"))
-    config.add(tk.nl())
-
+def create_service_toml_config(header_comment: str = None,
+                               id = None, name=None, description=None,
+                               host=None, port=None, serve_bootstrap=None,
+                               fixed_sturdy_ref_token=None,
+                               reg_sturdy_ref=None, reg_category=None,
+                               resolver_sturdy_ref=None, resolver_alias=None) -> tk.TOMLDocument:
+    doc = tk.document()
+    doc.add(tk.comment(header_comment if header_comment else "MAS service configuration file"))
+    doc.add(tk.nl())
+
+    services = tk.aot()
     service = tk.table()
     service["id"] = id if id else str(uuid.uuid4())
     service["name"] = name if name else f"Service {service['id']}"
     service["description"] = description if description else f"No description for service {service['id']}"
-    config.add("service", service)
-
-    caps = tk.table()
-    caps["fixed_sturdy_ref_token"] = sturdy_ref_token if sturdy_ref_token else "a_sturdy_ref_token"
-    register_at = tk.aot()
-    register_at.append(tk.item({
-        "sturdy_ref": reg_sturdy_ref if reg_sturdy_ref else "capnp://the_host_key@host:port/a_sturdy_ref_token",
-        "category": reg_category if reg_category else "a registry category",
-    }))
-    config.add("register_at", register_at)
-    config.add("caps", caps)
-
-    network = tk.table()
-    network["host"] = host if host else "localhost"
-    network["port"] = port if port else 0
-    network["serve_bootstrap"] = serve_bootstrap if serve_bootstrap else True
-    config.add("network", network)
-
-    return config
-
+    service["fixed_sturdy_ref_token"] = fixed_sturdy_ref_token if fixed_sturdy_ref_token else "a_sturdy_ref_token"
+    services.append(service)
+
+    regs = tk.aot()
+    regs.name = "services.registries"
+    reg = tk.table()
+    reg["sturdy_ref"] = reg_sturdy_ref if reg_sturdy_ref else "capnp://the_host_key@host:port/a_sturdy_ref_token"
+    reg["category"] = reg_category if reg_category else "a registry category"
+    regs.append(reg)
+
+    doc.add("services", services)
+
+    resolvers = tk.aot()
+    resolvers.name = "resolvers"
+    res = tk.table()
+    res["alias"] = resolver_alias if resolver_alias else "some_alias_for_the_resolver"
+    res["resolver_sturdy_ref"] = reg_category if reg_category else "capnp://the_host_key@host:port/a_sturdy_ref_token"
+    resolvers.append(res)
+
+#    register_at = tk.aot()
+#    register_at.append(tk.item({
+#        "sturdy_ref": reg_sturdy_ref if reg_sturdy_ref else "capnp://the_host_key@host:port/a_sturdy_ref_token",
+#        "category": reg_category if reg_category else "a registry category",
+#    }))
+#    config.add("register_at", register_at)
+#    config.add("caps", caps)
+
+    vat = tk.table()
+    vat["host"] = host if host else "localhost"
+    vat["port"] = port if port else 0
+    vat["serve_bootstrap"] = serve_bootstrap if serve_bootstrap else True
+    doc.add("vat", vat)
+
+    return doc
 
 
 # def sign_sr_token_by_sk_and_encode_base64(self, sr_token):
@@ -194,6 +212,15 @@ def set_vat_id_from_sign_pk(self):
             int.from_bytes(self._sign_pk[24:32], byteorder=sys.byteorder, signed=False),
         ]
 
+    @property
+    def base64_vat_id(self):
+        return base64.urlsafe_b64encode(self._sign_pk).decode("utf-8")
+
+    def signature_of_vat_id(self):
+        public_key_bytes = base64.urlsafe_b64encode(self._sign_pk)
+        signature = pysodium.crypto_sign_detached(public_key_bytes, self._sign_sk)
+        return signature
+
     @property
     def storage_container(self):
         return self._storage_container
@@ -574,6 +601,10 @@ def __init__(self, restorer=None):
         self._connections = {}
         self._restorer = restorer if restorer else Restorer()
 
+    @property
+    def restorer(self):
+        return self._restorer
+
     async def connect(self, sturdy_ref, cast_as=None):
         if not sturdy_ref:
             return None

zalfmas_common/service.py

@@ -20,6 +20,8 @@
 import tomlkit as tk
 from zalfmas_common import common
 import zalfmas_capnp_schemas
+from zalfmas_common.common import ConnectionManager
+
 sys.path.append(os.path.dirname(zalfmas_capnp_schemas.__file__))
 import common_capnp
 import fbp_capnp
@@ -110,71 +112,98 @@ async def updateIdentity(self, oldId, newInfo, **kwargs):  # updateIdentity @4 (
                 s.description = newInfo.description
 
 
-async def init_and_run_service(name_to_service, host=None, port=0, serve_bootstrap=True, restorer=None,
-                               conn_man=None, name_to_service_srs=None, run_before_enter_eventloop=None,
-                               restorer_container_sr=None, read_from_stdin=False, **kwargs):
-    port = port if port else 0
+async def register_services(con_man: ConnectionManager,
+                            name_to_service: dict,
+                            admin: Admin,
+                            registries: list[dict]):
+    for name, cap in name_to_service.items():
+        for reg in registries:
+            try:
+                reg_sr = data["sturdy_ref"]
+                reg_name = data.get("name", "")
+                reg_cat_id = data.get("category_id", "")
+                print("Trying to register service with name:", reg_name, "@ category:", reg_cat_id)
+                registrar = await con_man.try_connect(reg_sr, cast_as=reg_capnp.Registrar)
+                if registrar:
+                    r = await registrar.register(cap=cap, regName=reg_name, categoryId=reg_cat_id)
+                    unreg_action = r.unreg
+                    rereg_sr = r.reregSR
+                    admin.store_unreg_data(name, unreg_action, rereg_sr)
+                    print("Registered service", name, "in category '", reg_cat_id, "' as '", reg_name, "'.")
+                else:
+                    print("Couldn't connect to registrar at sturdy_ref:", reg_sr)
+            except Exception as e:
+                print("Error registering service name:", name, "using data:", reg, ". Exception:", e)
+
 
-    # check for sturdy ref inputs
-    reg_config = {}
-    if read_from_stdin and not sys.stdin.isatty():
+async def register_vat_at_resolvers(con_man: ConnectionManager, resolvers: list):
+    for res in resolvers:
         try:
-            reg_config = json.loads(sys.stdin.read())
-            # print("read from stdin:", reg_config)
+            sr = res["sturdy_ref"]
+            print("Trying to register vat at resolver sturdy_ref:", sr)
+            registrar = await con_man.try_connect(sr, cast_as=persistence_capnp.HostPortResolver.Registrar)
+            if registrar:
+                req = registrar.register_request()
+                req.host = con_man.restorer.host
+                req.port = con_man.restorer.port
+                req.base64VatId = con_man.restorer.base64_vat_id
+                req.identityProof = con_man.restorer.signature_of_vat_id()
+                if "alias" in res:
+                    req.alias = res["alias"]
+                r = await req.send()
+                #r = await registrar.register(cap=name_to_service[name], regName=service_name,
+                #                             categoryId=alias)
+                hb = r.heartbeat
+                hb_int = r.secsHeartbeatInterval
+                capnp.getTimer().after_delay(hb_int * 10 ** 9).then(lambda: exit(0))
+                #admin.store_unreg_data(name, unreg_action, rereg_sr)
+                print("Registered at resolver vat with vat_id:", con_man.restorer.base64VatIdservice_name,
+                      f"and alias {res['alias']}." if "alias" in res else ".")
+            else:
+                print("Couldn't connect to registrar at sturdy_ref:", sr)
         except Exception as e:
-            print("service.py: Error reading from sys.stdin. Exception:", e)
+            print("Error registering vat. Exception:", e)
+
+
+async def init_and_run_service(name_to_service,
+                               host=None, port=0,
+                               serve_bootstrap=True,
+                               restorer=None,
+                               con_man=None,
+                               name_to_service_srs=None,
+                               run_before_enter_eventloop=None,
+                               restorer_container_sr=None,
+                               registries=None,
+                               resolvers=None):
+    port = port if port else 0
+
+    registries = registries if registries else []
+    resolvers = resolvers if resolvers else []
 
     if not restorer:
         restorer = common.Restorer()
-    if not conn_man:
-        conn_man = common.ConnectionManager(restorer)
+    if not con_man:
+        con_man = common.ConnectionManager(restorer)
     if not name_to_service_srs:
         name_to_service_srs = {}
 
     if restorer and restorer_container_sr:
-        restorer_container = await conn_man.try_connect(restorer_container_sr, cast_as=storage_capnp.Store.Container)
+        restorer_container = await con_man.try_connect(restorer_container_sr, cast_as=storage_capnp.Store.Container)
         if restorer_container:
             restorer.storage_container = restorer_container
             await restorer.init_vat_id_from_container()
             if not port:
                 await restorer.init_port_from_container()
                 port = restorer.port
 
-    # create and register admin with services
+    # create and register admin interface with services
     admin = Admin(list(name_to_service.values()))
     for s in name_to_service.values():
         if isinstance(s, AdministrableService):
             s.admin = admin
     if "admin" not in name_to_service and admin not in name_to_service.values():
         name_to_service["admin"] = admin
 
-    async def register_services(conn_man, admin, reg_config):
-        for name, data in reg_config.items():
-            try:
-                if isinstance(data, dict):
-                    reg_sr = data["reg_sr"]
-                    reg_name = data.get("reg_name", "")
-                    reg_cat_id = data.get("cat_id", "")
-                elif isinstance(data, str):
-                    reg_sr = data
-                    reg_name = ""
-                    reg_cat_id = ""
-                else:
-                    continue
-                print("trying to register name:", name, "data:", data)
-                registrar = await conn_man.try_connect(reg_sr, cast_as=reg_capnp.Registrar)
-                if registrar and name in name_to_service:
-                    r = await registrar.register(cap=name_to_service[name], regName=reg_name,
-                                                 categoryId=reg_cat_id)
-                    unreg_action = r.unreg
-                    rereg_sr = r.reregSR
-                    admin.store_unreg_data(name, unreg_action, rereg_sr)
-                    print("Registered", name, "in category '", reg_cat_id, "' as '", reg_name, "'.")
-                else:
-                    print("Couldn't connect to registrar at sturdy_ref:", reg_sr)
-            except Exception as e:
-                print("Error registering service name:", name, ". Exception:", e)
-
     async def new_connection(stream):
         await capnp.TwoPartyServer(stream, bootstrap=restorer).on_disconnect()
 
@@ -183,24 +212,25 @@ async def new_connection(stream):
         restorer.port = server.sockets[0].getsockname()[1]
 
         for name, s in name_to_service.items():
-            res = await restorer.save_str(s, name_to_service_srs.get(name, None))
+            res = await restorer.save_str(cap=s, fixed_sr_token=name_to_service_srs.get(name, None))
             name_to_service_srs[name] = res["sturdy_ref"]
             print("service:", name, "sr:", res["sturdy_ref"])
         print("restorer_sr:", restorer.sturdy_ref_str())
 
-        await register_services(conn_man, admin, reg_config)
+        #await register_services(con_man, name_to_service, admin, registries)
+        await register_vat_at_resolvers(con_man, resolvers)
         if run_before_enter_eventloop:
             run_before_enter_eventloop()
         async with server:
             await server.serve_forever()
     #else:
-    #    await register_services(conn_man, admin, reg_config)
+    #    await register_services(con_man, admin, reg_config)
     #    if run_before_enter_eventloop:
     #        run_before_enter_eventloop()
-    #    await conn_man.manage_forever()
+    #    await con_man.manage_forever()
 
 
-def handle_default_service_args(parser, config: dict=None):
+def handle_default_service_args_with_dict(parser, config: dict=None):
     args = parser.parse_args()
 
     remove_keys = []
@@ -233,13 +263,62 @@ def handle_default_service_args(parser, config: dict=None):
         with open(args.config_toml, "r") as f:
             toml_config = tk.load(f)
         config.update({ k:v for k, v in toml_config.items() if type(v) is not tk.items.Table})
-        config.update(toml_config.get("defaults", {}))
+        if "vat" in toml_config:
+            config.update({f"vat.{k}": v for k, v in toml_config["vat"].items() if type(v) is not tk.items.Table})
     else:
         parser.error("argument config_toml: expected path to config TOML file")
 
     for k in remove_keys:
         del config[k]
-    return config, args
+    return config, args, toml_config
+
+def handle_default_service_args_toml(parser, template_toml_str: str=None):
+    args = parser.parse_args()
+
+    if args.output_toml_config:
+        print(template_toml_str)
+        exit(0)
+    elif args.write_toml_config:
+        with open(args.write_toml_config, "w") as _:
+            _.write(template_toml_str)
+            exit(0)
+    elif args.config_toml is not None:
+        with open(args.config_toml, "r") as f:
+            toml_config = tk.load(f)
+    else:
+        parser.error("argument config_toml: expected path to config TOML file")
+
+    return toml_config, args
+
+
+def handle_default_service_args(parser, path_to_template_config_json=None, path_to_service_py=None,
+                                relative_path_from_service_py_to_default_configs_folder="default_configs"):
+    args = parser.parse_args()
+    json_config = {}
+    template_config_json = {}
+    if not path_to_template_config_json and path_to_service_py:
+        path_to_template_config_json = (
+            os.path.join(os.path.dirname(path_to_service_py),
+                         relative_path_from_service_py_to_default_configs_folder,
+                         os.path.basename(path_to_service_py).replace(".py", ".json")))
+    if path_to_template_config_json:
+        with open(path_to_template_config_json, "r") as f:
+            template_config_json = json.load(f)
+
+    if args.output_toml_config:
+        print(json.dumps(template_config_json))
+        exit(0)
+    elif args.write_toml_config:
+        with open(args.write_toml_config, "w") as _:
+            _.write(json.dumps(template_config_json))
+            exit(0)
+    elif args.config_toml is not None:
+        with open(args.config_toml, "r") as f:
+            json_config = json.load(f)
+    else:
+        parser.error("argument config_toml: expected path to config JSON file")
+
+    return json_config, args
 
 
 def create_default_args_parser(