remote/coordinator: add CreatePlace RPC

labgrid/remote/client.py

@@ -32,6 +32,11 @@
 # TODO: drop if Python >= 3.11 guaranteed
 from exceptiongroup import ExceptionGroup  # pylint: disable=redefined-builtin
 
+from labgrid.remote.grpc.interceptor.client import (
+    IdentityClientStreamStreamInterceptor,
+    IdentityClientUnaryUnaryInterceptor,
+)
+
 from .common import (
     ResourceEntry,
     ResourceMatch,
@@ -120,9 +125,19 @@ def __attrs_post_init__(self):
             ("grpc.http2.max_pings_without_data", 0),  # no limit
         ]
 
+        identity = {
+            "username": self.getuser(),
+            "hostname": self.gethostname(),
+            "user_agent": f"labgrid-client {labgrid_version()}",
+        }
+        interceptors = [
+            IdentityClientUnaryUnaryInterceptor(**identity),
+            IdentityClientStreamStreamInterceptor(**identity),
+        ]
         self.channel = grpc.aio.insecure_channel(
             target=self.address,
             options=channel_options,
+            interceptors=interceptors,
         )
         self.stub = labgrid_coordinator_pb2_grpc.CoordinatorStub(self.channel)
 
@@ -139,10 +154,6 @@ async def start(self):
 
         self.pump_task = self.loop.create_task(self.message_pump())
         msg = labgrid_coordinator_pb2.ClientInMessage()
-        msg.startup.version = labgrid_version()
-        msg.startup.name = f"{self.gethostname()}/{self.getuser()}"
-        self.out_queue.put_nowait(msg)
-        msg = labgrid_coordinator_pb2.ClientInMessage()
         msg.subscribe.all_places = True
         self.out_queue.put_nowait(msg)
         msg = labgrid_coordinator_pb2.ClientInMessage()
@@ -555,15 +566,15 @@ async def print_place(self):
                         print(f"Matching resource '{name}' ({exporter}/{group_name}/{resource.cls}/{resource_name}):")  # pylint: disable=line-too-long
                         print(indent(pformat(resource.asdict()), prefix="  "))
 
-    async def add_place(self):
-        """Add a place to the coordinator"""
+    async def create_place(self):
+        """Create a place on the coordinator"""
         name = self.args.place
         if not name:
             raise UserError("missing place name. Set with -p <place> or via env var LG_PLACE")
 
-        request = labgrid_coordinator_pb2.AddPlaceRequest(name=name)
+        request = labgrid_coordinator_pb2.CreatePlaceRequest(name=name)
         try:
-            await self.stub.AddPlace(request)
+            await self.stub.CreatePlace(request)
             await self.sync_with_coordinator()
         except grpc.aio.AioRpcError as e:
             raise ServerError(e.details())
@@ -836,6 +847,21 @@ async def allow(self):
 
         print(f"allowed {self.args.user} for place {place.name}")
 
+    async def unshare(self):
+        """Remove another user's access to a previously acquired place"""
+        place = self.get_acquired_place()
+        if "/" not in self.args.user:
+            raise UserError(f"user {self.args.user} must be in <host>/<username> format")
+        request = labgrid_coordinator_pb2.UnsharePlaceRequest(name=place.name, user=self.args.user)
+
+        try:
+            await self.stub.UnsharePlace(request)
+            await self.sync_with_coordinator()
+        except grpc.aio.AioRpcError as e:
+            raise ServerError(e.details())
+
+        print(f"unshared place {place.name} with {self.args.user}")
+
     def get_target_resources(self, place):
         self._check_allowed(place)
         resources = {}
@@ -1917,9 +1943,9 @@ def get_parser(auto_doc_mode=False) -> "argparse.ArgumentParser | AutoProgramArg
 
     subparser = subparsers.add_parser(
         "create",
-        help="add a new place with the name specified via --place or the LG_PLACE environment variable",
+        help="create a new place with the name specified via --place or the LG_PLACE environment variable",
     )
-    subparser.set_defaults(func=ClientSession.add_place)
+    subparser.set_defaults(func=ClientSession.create_place)
 
     subparser = subparsers.add_parser("delete", help="delete an existing place")
     subparser.set_defaults(func=ClientSession.del_place)
@@ -1978,6 +2004,10 @@ def get_parser(auto_doc_mode=False) -> "argparse.ArgumentParser | AutoProgramArg
     subparser.add_argument("user", help="<host>/<username>")
     subparser.set_defaults(func=ClientSession.allow)
 
+    subparser = subparsers.add_parser("unshare", help="remove another user's access to a place")
+    subparser.add_argument("user", help="<host>/<username>")
+    subparser.set_defaults(func=ClientSession.unshare)
+
     subparser = subparsers.add_parser("env", help="generate a labgrid environment file for a place")
     subparser.set_defaults(func=ClientSession.print_env)
 

labgrid/remote/common.py

@@ -7,6 +7,8 @@
 import logging
 from datetime import datetime
 from fnmatch import fnmatchcase
+from typing import Optional
+import warnings
 
 import attr
 
@@ -481,6 +483,21 @@ def from_pb2(cls, pb2: labgrid_coordinator_pb2.Reservation):
         )
 
 
+def get_metadata_single_value_by_key(metadata, key: str) -> Optional[str]:
+    """Look up a single value by key in a metadata sequence of (key, value) pairs."""
+    values = [v for k, v in metadata or () if k == key]
+
+    if not values:
+        return None
+
+    if len(values) > 1:
+        warnings.warn(
+            "Multiple metadata KV pairs with the same key. The value of the first matching KV pair will be returned."
+        )
+
+    return values[0]
+
+
 async def queue_as_aiter(q):
     try:
         while True:

labgrid/remote/coordinator.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 import argparse
+import contextvars
 import logging
 import asyncio
 import traceback
@@ -10,11 +11,15 @@
 import copy
 import random
 import signal
+from typing import Optional
 
 import attr
 import grpc
 from grpc_reflection.v1alpha import reflection
 
+from labgrid.remote.grpc.interceptor.server import IdentityServerInterceptor
+from labgrid.remote.identity import ClientIdentity, infer_peer_identity
+
 from .common import (
     ResourceEntry,
     ResourceMatch,
@@ -30,6 +35,10 @@
 from .generated import labgrid_coordinator_pb2_grpc
 from ..util import atomic_replace, labgrid_version, yaml, Timeout
 
+client_identity_context: contextvars.ContextVar[Optional[ClientIdentity]] = contextvars.ContextVar(
+    "client_identity", default=None
+)
+
 
 @contextmanager
 def warn_if_slow(prefix, *, level=logging.WARNING, limit=0.1):
@@ -317,9 +326,17 @@ async def ClientStream(self, request_iterator, context):
         assert peer not in self.clients
         out_msg_queue = asyncio.Queue()
 
+        identity = client_identity_context.get()
+        if identity:
+            logging.debug("client identity provided in gRPC metadata")
+            logging.debug(identity)
+            self.clients[peer] = ClientSession(self, peer, identity.id, out_msg_queue, identity.user_agent)
+
         async def request_task():
             name = None
             version = None
+            if peer in self.clients:
+                session = self.clients[peer]
             try:
                 async for in_msg in request_iterator:
                     in_msg: labgrid_coordinator_pb2.ClientInMessage
@@ -330,6 +347,9 @@ async def request_task():
                         out_msg.sync.id = in_msg.sync.id
                         out_msg_queue.put_nowait(out_msg)
                     elif kind == "startup":
+                        if peer in self.clients:
+                            logging.debug("already setup, probably because identity was provided in metadata")
+                            continue
                         version = in_msg.startup.version
                         name = in_msg.startup.name
                         session = self.clients[peer] = ClientSession(self, peer, name, out_msg_queue, version)
@@ -409,9 +429,23 @@ async def ExporterStream(self, request_iterator, context):
         out_msg.hello.version = labgrid_version()
         yield out_msg
 
+        identity = client_identity_context.get()
+        if identity:
+            logging.debug("exporter identity provided in gRPC metadata")
+            logging.debug(identity)
+            if existing := self.get_exporter_by_name(identity.id):
+                await context.abort(
+                    grpc.StatusCode.ALREADY_EXISTS,
+                    f"startup failed: exporter with name '{identity.id}' is already connected from {existing.peer}",
+                )
+            self.exporters[peer] = ExporterSession(self, peer, identity.id, command_queue, identity.user_agent)
+            startup_done.set()
+
         async def request_task():
             name = None
             version = None
+            if peer in self.exporters:
+                session = self.exporters[peer]
             try:
                 async for in_msg in request_iterator:
                     in_msg: labgrid_coordinator_pb2.ExporterInMessage
@@ -422,6 +456,9 @@ async def request_task():
                         cmd.complete(in_msg.response)
                         logging.debug("Command %s is done", cmd)
                     elif kind == "startup":
+                        if peer in self.exporters:
+                            logging.debug("already setup, probably because identity was provided in metadata")
+                            continue
                         version = in_msg.startup.version
                         name = in_msg.startup.name
                         if existing := self.get_exporter_by_name(name):
@@ -499,8 +536,11 @@ async def request_task():
             except KeyError:
                 logging.info("Never received startup from peer %s that disconnected", peer)
 
-    @locked
-    async def AddPlace(self, request, context):
+    async def create_place(
+        self,
+        request: labgrid_coordinator_pb2.CreatePlaceRequest | labgrid_coordinator_pb2.AddPlaceRequest,
+        context,
+    ) -> Place:
         name = request.name
         if not name or not isinstance(name, str):
             await context.abort(grpc.StatusCode.INVALID_ARGUMENT, "name was not a string")
@@ -511,8 +551,19 @@ async def AddPlace(self, request, context):
         self.places[name] = place
         self._publish_place(place)
         self.save_later()
+        return place
+
+    @locked
+    async def AddPlace(self, request, context):
+        await self.create_place(request, context)
         return labgrid_coordinator_pb2.AddPlaceResponse()
 
+    @locked
+    async def CreatePlace(self, request, context):
+        logging.debug("CreatePlace name=%s", request.name)
+        place = await self.create_place(request, context)
+        return labgrid_coordinator_pb2.CreatePlaceResponse(place=place.as_pb2())
+
     @locked
     async def DeletePlace(self, request, context):
         name = request.name
@@ -852,7 +903,7 @@ async def AcquirePlace(self, request, context):
         peer = context.peer()
         name = request.placename
         try:
-            username = self.clients[peer].name
+            username = infer_peer_identity(self.clients, context, client_identity_context)
         except KeyError:
             await context.abort(grpc.StatusCode.FAILED_PRECONDITION, f"Peer {peer} does not have a valid session")
         print(request)
@@ -922,7 +973,7 @@ async def AllowPlace(self, request, context):
         user = request.user
         peer = context.peer()
         try:
-            username = self.clients[peer].name
+            username = infer_peer_identity(self.clients, context, client_identity_context)
         except KeyError:
             await context.abort(grpc.StatusCode.FAILED_PRECONDITION, f"Peer {peer} does not have a valid session")
         try:
@@ -941,9 +992,55 @@ async def AllowPlace(self, request, context):
         self.save_later()
         return labgrid_coordinator_pb2.AllowPlaceResponse()
 
+    @locked
+    async def UnsharePlace(self, request, context):
+        logging.debug("UnsharePlace name=%s user=%s", request.name, request.user)
+
+        placename = request.name
+        user = request.user
+        peer = context.peer()
+        try:
+            username = infer_peer_identity(self.clients, context, client_identity_context)
+        except KeyError:
+            await context.abort(grpc.StatusCode.FAILED_PRECONDITION, f"Peer {peer} does not have a valid session")
+        try:
+            place = self.places[placename]
+        except KeyError:
+            await context.abort(grpc.StatusCode.INVALID_ARGUMENT, f"Place {placename} does not exist")
+        if not place.acquired:
+            await context.abort(grpc.StatusCode.FAILED_PRECONDITION, f"Place {placename} is not acquired")
+        if not place.acquired == username:
+            await context.abort(
+                grpc.StatusCode.FAILED_PRECONDITION, f"Place {placename} is not acquired by {username}"
+            )
+        try:
+            place.allowed.remove(user)
+            place.touch()
+            self._publish_place(place)
+            self.save_later()
+        except KeyError:
+            await context.abort(grpc.StatusCode.FAILED_PRECONDITION, f"Place {placename} is not shared with {user}")
+
+        return labgrid_coordinator_pb2.UnsharePlaceResponse()
+
     def _get_places(self):
         return {k: v.asdict() for k, v in self.places.items()}
 
+    @locked
+    async def GetPlace(self, request, context):
+        name = request.name
+        logging.debug("GetPlace name=%s", name)
+        if not name or not isinstance(name, str):
+            await context.abort(grpc.StatusCode.INVALID_ARGUMENT, "name was not a string")
+        try:
+            place = self.places[name]
+            return labgrid_coordinator_pb2.GetPlaceResponse(place=place.as_pb2())
+        except KeyError:
+            await context.abort(grpc.StatusCode.INVALID_ARGUMENT, f"Place {name} does not exist")
+        except Exception:
+            logging.exception("error during get place")
+            await context.abort(grpc.StatusCode.INTERNAL, "internal error")
+
     @locked
     async def GetPlaces(self, unused_request, unused_context):
         logging.debug("GetPlaces")
@@ -1076,7 +1173,10 @@ async def CreateReservation(self, request: labgrid_coordinator_pb2.CreateReserva
                     await context.abort(grpc.StatusCode.INVALID_ARGUMENT, f"Value {v} is invalid")
                 fltr[k] = v
 
-        owner = self.clients[peer].name
+        try:
+            owner = infer_peer_identity(self.clients, context, client_identity_context)
+        except KeyError:
+            await context.abort(grpc.StatusCode.FAILED_PRECONDITION, f"Peer {peer} does not have a valid session")
         res = Reservation(owner=owner, prio=request.prio, filters=fltrs)
         self.reservations[res.token] = res
         self.schedule_reservations()
@@ -1108,7 +1208,6 @@ async def GetReservations(self, request: labgrid_coordinator_pb2.GetReservations
         reservations = [x.as_pb2() for x in self.reservations.values()]
         return labgrid_coordinator_pb2.GetReservationsResponse(reservations=reservations)
 
-
 async def serve(listen, cleanup) -> None:
     asyncio.current_task().set_name("coordinator-serve")
     # It seems since https://github.com/grpc/grpc/pull/34647, the
@@ -1126,6 +1225,7 @@ async def serve(listen, cleanup) -> None:
     ]
     server = grpc.aio.server(
         options=channel_options,
+        interceptors=[IdentityServerInterceptor(client_identity_context)],
     )
     coordinator = Coordinator()
     labgrid_coordinator_pb2_grpc.add_CoordinatorServicer_to_server(coordinator, server)