Pass extra session info to subscription callbacks that requested so

* For some niche use cases, during the execution of a subscription callback,
  it is useful to have access to certain session information (e.g., username
  and NETCONF session ID that triggered the callback, etc.)

* Add "extra_info" argument to all "SysrepoSession.subscribe_xxx" methods
  (with a default value of False) to indicate the desire to receive extra
  keyword arguments during the callback call

* Pass "netconf_id" (int) and "user" (str) keyword arguments to all callbacks
  which were subscribed with "extra_info=True"

* Extend unit tests to include and cover this new feature

Author: nicolasmorini

cffi/cdefs.h

@@ -85,6 +85,8 @@ int sr_session_stop(sr_session_ctx_t *);
 int sr_session_switch_ds(sr_session_ctx_t *, sr_datastore_t);
 sr_datastore_t sr_session_get_ds(sr_session_ctx_t *);
 sr_conn_ctx_t *sr_session_get_connection(sr_session_ctx_t *);
+uint32_t sr_session_get_event_nc_id(sr_session_ctx_t *);
+const char *sr_session_get_event_user(sr_session_ctx_t *);
 int sr_get_error(sr_session_ctx_t *, const sr_error_info_t **);
 int sr_set_error(sr_session_ctx_t *, const char *, const char *, ...);
 

sysrepo/session.py

@@ -121,6 +121,30 @@ def set_error(self, xpath: Optional[str], message: str):
             lib.sr_set_error, self.cdata, str2c(xpath), str2c("%s"), str2c(message)
         )
 
+    def get_netconf_id(self) -> int:
+        """
+        It can only be called on an implicit sysrepo.Session (i.e., it can only be
+        called from an event callback)
+
+        :returns: the NETCONF session ID set for the event originator sysrepo session
+        """
+        if not self.is_implicit:
+            raise SysrepoUnsupportedError(
+                "can only report netconf id on implicit sessions"
+            )
+        return lib.sr_session_get_event_nc_id(self.cdata)
+
+    def get_user(self) -> str:
+        """
+        It can only be called on an implicit sysrepo.Session (i.e., it can only be
+        called from an event callback)
+
+        :returns: the effective username of the event originator sysrepo session
+        """
+        if not self.is_implicit:
+            raise SysrepoUnsupportedError("can only report user on implicit sessions")
+        return c2str(lib.sr_session_get_event_user(self.cdata))
+
     def get_ly_ctx(self) -> libyang.Context:
         """
         :returns:
@@ -152,6 +176,13 @@ def get_ly_ctx(self) -> libyang.Context:
         have changed.
     :arg private_data:
         Private context opaque to sysrepo used when subscribing.
+    :arg kwargs (optional):
+        If the callback was registered with the argument extra_info=True (see
+        Session.subscribe_module_change), then extra keyword arguments are passed when
+        calling the callback:
+            * netconf_id: the NETCONF session ID set for the event originator
+                sysrepo session
+            * user: the effective username of the event originator sysrepo session
 
     When event is one of ("update", "change"), if the callback raises an exception, the
     changes will be rejected and the error will be forwarded to the client that made the
@@ -174,7 +205,8 @@ def subscribe_module_change(
         enabled: bool = False,
         private_data: Any = None,
         asyncio_register: bool = False,
-        include_implicit_defaults: bool = True
+        include_implicit_defaults: bool = True,
+        extra_info: bool = False
     ) -> None:
         """
         Subscribe for changes made in the specified module.
@@ -210,6 +242,10 @@ def subscribe_module_change(
             monitored read file descriptors. Implies `no_thread=True`.
         :arg include_implicit_defaults:
             Include implicit default nodes in changes.
+        :arg extra_info:
+            When True, the given callback is called with extra keyword arguments
+            containing extra information of the sysrepo session that gave origin to the
+            event (see ModuleChangeCallbackType for more details)
         """
         if self.is_implicit:
             raise SysrepoUnsupportedError("cannot subscribe with implicit sessions")
@@ -220,6 +256,7 @@ def subscribe_module_change(
             private_data,
             asyncio_register=asyncio_register,
             include_implicit_defaults=include_implicit_defaults,
+            extra_info=extra_info,
         )
         sub_p = ffi.new("sr_subscription_ctx_t **")
 
@@ -253,6 +290,13 @@ def subscribe_module_change(
         module operational data.
     :arg private_data:
         Private context opaque to sysrepo used when subscribing.
+    :arg kwargs (optional):
+        If the callback was registered with the argument extra_info=True (see
+        Session.subscribe_module_change), then extra keyword arguments are passed when
+        calling the callback:
+            * netconf_id: the NETCONF session ID set for the event originator
+                sysrepo session
+            * user: the effective username of the event originator sysrepo session
 
     The callback is expected to return a python dictionary containing the operational
     data. The dictionary should be in the libyang "dict" format. It will be parsed to a
@@ -272,7 +316,8 @@ def subscribe_oper_data_request(
         no_thread: bool = False,
         private_data: Any = None,
         asyncio_register: bool = False,
-        strict: bool = False
+        strict: bool = False,
+        extra_info: bool = False
     ) -> None:
         """
         Register for providing operational data at the given xpath.
@@ -296,13 +341,21 @@ def subscribe_oper_data_request(
         :arg strict:
             Reject the whole data returned by callback if it contains elements without
             schema definition.
+        :arg extra_info:
+            When True, the given callback is called with extra keyword arguments
+            containing extra information of the sysrepo session that gave origin to the
+            event (see OperDataCallbackType for more details)
         """
         if self.is_implicit:
             raise SysrepoUnsupportedError("cannot subscribe with implicit sessions")
         _check_subscription_callback(callback, self.OperDataCallbackType)
 
         sub = Subscription(
-            callback, private_data, asyncio_register=asyncio_register, strict=strict
+            callback,
+            private_data,
+            asyncio_register=asyncio_register,
+            strict=strict,
+            extra_info=extra_info,
         )
         sub_p = ffi.new("sr_subscription_ctx_t **")
 
@@ -369,6 +422,13 @@ def subscribe_oper_data_request(
         will be called with 'abort'.
     :arg private_data:
         Private context opaque to sysrepo used when subscribing.
+    :arg kwargs (optional):
+        If the callback was registered with the argument extra_info=True (see
+        Session.subscribe_module_change), then extra keyword arguments are passed when
+        calling the callback:
+            * netconf_id: the NETCONF session ID set for the event originator
+                sysrepo session
+            * user: the effective username of the event originator sysrepo session
 
     The callback is expected to return a python dictionary containing the RPC output
     data. The dictionary should be in the libyang "dict" format and must only contain
@@ -393,7 +453,8 @@ def subscribe_rpc_call(
         private_data: Any = None,
         asyncio_register: bool = False,
         strict: bool = False,
-        include_implicit_defaults: bool = True
+        include_implicit_defaults: bool = True,
+        extra_info: bool = False
     ) -> None:
         """
         Subscribe for the delivery of an RPC/action.
@@ -418,6 +479,10 @@ def subscribe_rpc_call(
             schema definition.
         :arg include_implicit_defaults:
             Include implicit defaults into input parameters passed to callbacks.
+        :arg extra_info:
+            When True, the given callback is called with extra keyword arguments
+            containing extra information of the sysrepo session that gave origin to the
+            event (see RpcCallbackType for more details)
         """
         if self.is_implicit:
             raise SysrepoUnsupportedError("cannot subscribe with implicit sessions")
@@ -429,6 +494,7 @@ def subscribe_rpc_call(
             asyncio_register=asyncio_register,
             strict=strict,
             include_implicit_defaults=include_implicit_defaults,
+            extra_info=extra_info,
         )
         sub_p = ffi.new("sr_subscription_ctx_t **")
 
@@ -480,6 +546,13 @@ def subscribe_rpc_call(
         Timestamp of the notification as an unsigned 32-bits integer.
     :arg private_data:
         Private context opaque to sysrepo used when subscribing.
+    :arg kwargs (optional):
+        If the callback was registered with the argument extra_info=True (see
+        Session.subscribe_module_change), then extra keyword arguments are passed when
+        calling the callback:
+            * netconf_id: the NETCONF session ID set for the event originator
+                sysrepo session
+            * user: the effective username of the event originator sysrepo session
     """
 
     def subscribe_notification(
@@ -492,7 +565,8 @@ def subscribe_notification(
         stop_time: int = 0,
         no_thread: bool = False,
         asyncio_register: bool = False,
-        private_data: Any = None
+        private_data: Any = None,
+        extra_info: bool = False
     ) -> None:
         """
         Subscribe for the delivery of a notification.
@@ -516,6 +590,10 @@ def subscribe_notification(
             read file descriptors. Implies no_thread=True.
         :arg private_data:
             Private context passed to the callback function, opaque to sysrepo.
+        :arg extra_info:
+            When True, the given callback is called with extra keyword arguments
+            containing extra information of the sysrepo session that gave origin to the
+            event (see RpcCallbackType for more details)
         """
 
         if self.is_implicit:
@@ -526,6 +604,7 @@ def subscribe_notification(
             callback,
             private_data,
             asyncio_register=asyncio_register,
+            extra_info=extra_info,
         )
 
         sub_p = ffi.new("sr_subscription_ctx_t **")

sysrepo/subscription.py

@@ -33,6 +33,7 @@ def __init__(
         asyncio_register: bool = False,
         strict: bool = False,
         include_implicit_defaults: bool = True,
+        extra_info: bool = False,
     ):
         """
         :arg callback:
@@ -49,6 +50,10 @@ def __init__(
         :arg include_implicit_defaults:
             If True, include implicit default nodes into Change objects passed to module
             change callbacks and into input parameters passed to RPC/action callbacks.
+        :arg extra_info:
+            When True, the given callback is called with extra keyword arguments
+            containing extra information of the sysrepo session that gave origin to the
+            event
         """
         if is_async_func(callback) and not asyncio_register:
             raise ValueError(
@@ -59,6 +64,7 @@ def __init__(
         self.asyncio_register = asyncio_register
         self.strict = strict
         self.include_implicit_defaults = include_implicit_defaults
+        self.extra_info = extra_info
         if asyncio_register:
             self.loop = asyncio.get_event_loop()
         else:
@@ -214,6 +220,13 @@ def module_change_callback(session, module, xpath, event, req_id, priv):
         callback = subscription.callback
         private_data = subscription.private_data
         event_name = EVENT_NAMES[event]
+        if subscription.extra_info:
+            extra_info = {
+                "netconf_id": session.get_netconf_id(),
+                "user": session.get_user(),
+            }
+        else:
+            extra_info = {}
 
         if is_async_func(callback):
             task_id = (event, req_id)
@@ -230,7 +243,7 @@ def module_change_callback(session, module, xpath, event, req_id, priv):
                     )
                 )
                 task = subscription.loop.create_task(
-                    callback(event_name, req_id, changes, private_data)
+                    callback(event_name, req_id, changes, private_data, **extra_info)
                 )
                 task.add_done_callback(
                     functools.partial(subscription.task_done, task_id, event_name)
@@ -257,7 +270,7 @@ def module_change_callback(session, module, xpath, event, req_id, priv):
                     include_implicit_defaults=subscription.include_implicit_defaults,
                 )
             )
-            callback(event_name, req_id, changes, private_data)
+            callback(event_name, req_id, changes, private_data, **extra_info)
 
         return lib.SR_ERR_OK
 
@@ -328,12 +341,21 @@ def oper_data_callback(session, module, xpath, req_xpath, req_id, parent, priv):
         subscription = ffi.from_handle(priv)
         callback = subscription.callback
         private_data = subscription.private_data
+        if subscription.extra_info:
+            extra_info = {
+                "netconf_id": session.get_netconf_id(),
+                "user": session.get_user(),
+            }
+        else:
+            extra_info = {}
 
         if is_async_func(callback):
             task_id = req_id
 
             if task_id not in subscription.tasks:
-                task = subscription.loop.create_task(callback(req_xpath, private_data))
+                task = subscription.loop.create_task(
+                    callback(req_xpath, private_data, **extra_info)
+                )
                 task.add_done_callback(
                     functools.partial(subscription.task_done, task_id, "oper")
                 )
@@ -349,7 +371,7 @@ def oper_data_callback(session, module, xpath, req_xpath, req_id, parent, priv):
             oper_data = task.result()
 
         else:
-            oper_data = callback(req_xpath, private_data)
+            oper_data = callback(req_xpath, private_data, **extra_info)
 
         if isinstance(oper_data, dict):
             # convert oper_data to a libyang.DNode object
@@ -438,13 +460,20 @@ def rpc_callback(session, xpath, input_node, event, req_id, output_node, priv):
                 ).values()
             )
         )
+        if subscription.extra_info:
+            extra_info = {
+                "netconf_id": session.get_netconf_id(),
+                "user": session.get_user(),
+            }
+        else:
+            extra_info = {}
 
         if is_async_func(callback):
             task_id = (event, req_id)
 
             if task_id not in subscription.tasks:
                 task = subscription.loop.create_task(
-                    callback(xpath, input_dict, event_name, private_data)
+                    callback(xpath, input_dict, event_name, private_data, **extra_info)
                 )
                 task.add_done_callback(
                     functools.partial(subscription.task_done, task_id, event_name)
@@ -461,7 +490,9 @@ def rpc_callback(session, xpath, input_node, event, req_id, output_node, priv):
             output_dict = task.result()
 
         else:
-            output_dict = callback(xpath, input_dict, event_name, private_data)
+            output_dict = callback(
+                xpath, input_dict, event_name, private_data, **extra_info
+            )
 
         if event != lib.SR_EV_RPC:
             # May happen when there are multiple callback registered for the
@@ -543,15 +574,27 @@ def event_notif_tree_callback(session, notif_type, notif, timestamp, priv):
                 ).values()
             )
         )
+        if subscription.extra_info:
+            extra_info = {
+                "netconf_id": session.get_netconf_id(),
+                "user": session.get_user(),
+            }
+        else:
+            extra_info = {}
+
         if is_async_func(callback):
             task = subscription.loop.create_task(
-                callback(xpath, notif_type, notif_dict, timestamp, private_data)
+                callback(
+                    xpath, notif_type, notif_dict, timestamp, private_data, **extra_info
+                )
             )
             task.add_done_callback(
                 functools.partial(subscription.task_done, None, "notif")
             )
         else:
-            callback(xpath, notif_type, notif_dict, timestamp, private_data)
+            callback(
+                xpath, notif_type, notif_dict, timestamp, private_data, **extra_info
+            )
 
     except BaseException:
         # ATTENTION: catch all exceptions!

tests/test_session.py

@@ -125,3 +125,11 @@ def iface(name, field):
                     }
                 },
             )
+
+    def test_get_netconf_id_and_get_user_are_only_available_in_implicit_session(self):
+        with self.conn.start_session("running") as sess:
+            with self.assertRaises(sysrepo.SysrepoUnsupportedError):
+                sess.get_netconf_id()
+
+            with self.assertRaises(sysrepo.SysrepoUnsupportedError):
+                sess.get_user()