grpc: add session crypto information support

sambacc/grpc/backend.py

@@ -16,7 +16,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>
 #
 
-from typing import Any, Union
+from typing import Any, Union, Optional
 
 import dataclasses
 import json
@@ -30,9 +30,22 @@
 
 @dataclasses.dataclass
 class Versions:
-    samba_version: str = "foo"
-    sambacc_version: str = "bar"
-    container_version: str = "baz"
+    samba_version: str = ""
+    sambacc_version: str = ""
+    container_version: str = ""
+
+
+@dataclasses.dataclass
+class SessionCrypto:
+    cipher: str
+    degree: str
+
+    @classmethod
+    def load(cls, json_object: dict[str, Any]) -> Self:
+        cipher = json_object.get("cipher", "")
+        cipher = "" if cipher == "-" else cipher
+        degree = json_object.get("degree", "")
+        return cls(cipher=cipher, degree=degree)
 
 
 @dataclasses.dataclass
@@ -45,9 +58,15 @@ class Session:
     session_dialect: str
     uid: int
     gid: int
+    encryption: Optional[SessionCrypto] = None
+    signing: Optional[SessionCrypto] = None
 
     @classmethod
     def load(cls, json_object: dict[str, Any]) -> Self:
+        _encryption = json_object.get("encryption")
+        encryption = SessionCrypto.load(_encryption) if _encryption else None
+        _signing = json_object.get("signing")
+        signing = SessionCrypto.load(_signing) if _signing else None
         return cls(
             session_id=json_object.get("session_id", ""),
             username=json_object.get("username", ""),
@@ -57,6 +76,8 @@ def load(cls, json_object: dict[str, Any]) -> Self:
             session_dialect=json_object.get("session_dialect", ""),
             uid=int(json_object.get("uid", -1)),
             gid=int(json_object.get("gid", -1)),
+            encryption=encryption,
+            signing=signing,
         )
 
 

sambacc/grpc/generated/control_pb2.pyi

@@ -100,6 +100,23 @@ class StatusRequest(google.protobuf.message.Message):
 
 global___StatusRequest = StatusRequest
 
+class SessionCrypto(google.protobuf.message.Message):
+    DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+    CIPHER_FIELD_NUMBER: builtins.int
+    DEGREE_FIELD_NUMBER: builtins.int
+    cipher: builtins.str
+    degree: builtins.str
+    def __init__(
+        self,
+        *,
+        cipher: builtins.str = ...,
+        degree: builtins.str = ...,
+    ) -> None: ...
+    def ClearField(self, field_name: typing_extensions.Literal["cipher", b"cipher", "degree", b"degree"]) -> None: ...
+
+global___SessionCrypto = SessionCrypto
+
 class SessionInfo(google.protobuf.message.Message):
     DESCRIPTOR: google.protobuf.descriptor.Descriptor
 
@@ -111,6 +128,8 @@ class SessionInfo(google.protobuf.message.Message):
     SESSION_DIALECT_FIELD_NUMBER: builtins.int
     UID_FIELD_NUMBER: builtins.int
     GID_FIELD_NUMBER: builtins.int
+    ENCRYPTION_FIELD_NUMBER: builtins.int
+    SIGNING_FIELD_NUMBER: builtins.int
     session_id: builtins.str
     username: builtins.str
     groupname: builtins.str
@@ -119,6 +138,10 @@ class SessionInfo(google.protobuf.message.Message):
     session_dialect: builtins.str
     uid: builtins.int
     gid: builtins.int
+    @property
+    def encryption(self) -> global___SessionCrypto: ...
+    @property
+    def signing(self) -> global___SessionCrypto: ...
     def __init__(
         self,
         *,
@@ -130,8 +153,11 @@ class SessionInfo(google.protobuf.message.Message):
         session_dialect: builtins.str = ...,
         uid: builtins.int = ...,
         gid: builtins.int = ...,
+        encryption: global___SessionCrypto | None = ...,
+        signing: global___SessionCrypto | None = ...,
     ) -> None: ...
-    def ClearField(self, field_name: typing_extensions.Literal["gid", b"gid", "groupname", b"groupname", "hostname", b"hostname", "remote_machine", b"remote_machine", "session_dialect", b"session_dialect", "session_id", b"session_id", "uid", b"uid", "username", b"username"]) -> None: ...
+    def HasField(self, field_name: typing_extensions.Literal["encryption", b"encryption", "signing", b"signing"]) -> builtins.bool: ...
+    def ClearField(self, field_name: typing_extensions.Literal["encryption", b"encryption", "gid", b"gid", "groupname", b"groupname", "hostname", b"hostname", "remote_machine", b"remote_machine", "session_dialect", b"session_dialect", "session_id", b"session_id", "signing", b"signing", "uid", b"uid", "username", b"username"]) -> None: ...
 
 global___SessionInfo = SessionInfo
 

sambacc/grpc/protobufs/control.proto

@@ -32,6 +32,11 @@ message GeneralInfo {
 
 message StatusRequest {}
 
+message SessionCrypto {
+    string cipher = 1;
+    string degree = 2;
+}
+
 message SessionInfo {
     string session_id = 1;
     string username = 2;
@@ -41,6 +46,8 @@ message SessionInfo {
     string session_dialect = 6;
     uint32 uid = 7;
     uint32 gid = 8;
+    SessionCrypto encryption = 9;
+    SessionCrypto signing = 10;
 }
 
 message ConnInfo {

sambacc/grpc/server.py

@@ -72,6 +72,14 @@ def _get_info(backend: Backend) -> pb.GeneralInfo:
     )
 
 
+def _convert_crypto(
+    crypto: Optional[rbe.SessionCrypto],
+) -> Optional[pb.SessionCrypto]:
+    if not crypto:
+        return None
+    return pb.SessionCrypto(cipher=crypto.cipher, degree=crypto.degree)
+
+
 def _convert_session(session: rbe.Session) -> pb.SessionInfo:
     info = pb.SessionInfo(
         session_id=session.session_id,
@@ -80,6 +88,8 @@ def _convert_session(session: rbe.Session) -> pb.SessionInfo:
         remote_machine=session.remote_machine,
         hostname=session.hostname,
         session_dialect=session.session_dialect,
+        encryption=_convert_crypto(session.encryption),
+        signing=_convert_crypto(session.signing),
     )
     # python side takes -1 to mean not found uid/gid. in protobufs
     # that would mean the fields are unset

tests/test_grpc_backend.py

@@ -156,6 +156,12 @@ def _status_json1_check(status):
     assert s1.uid == 103107
     assert s1.gid == 102513
     assert len(status.tcons) == 1
+    assert s1.encryption
+    assert s1.encryption.cipher == ""
+    assert s1.encryption.degree == "none"
+    assert s1.signing
+    assert s1.signing.cipher == "AES-128-GMAC"
+    assert s1.signing.degree == "partial"
     t1 = status.tcons[0]
     assert t1.tcon_id == "3757739897"
     assert t1.session_id == "2891148582"

tests/test_grpc_server.py

@@ -194,7 +194,18 @@ def test_status(mock_grpc_server):
 
     assert mock_grpc_server.backend._counter["get_status"] == 1
     assert rsp.server_timestamp == "2025-05-08T20:41:57.273489+0000"
-    # TODO data assertions
+    # data assertions
+    assert len(rsp.sessions) == 1
+    assert rsp.sessions[0].session_id == "2891148582"
+    assert rsp.sessions[0].uid == 103107
+    assert rsp.sessions[0].gid == 102513
+    assert rsp.sessions[0].username == "DOMAIN1\\bwayne"
+    assert rsp.sessions[0].encryption
+    assert rsp.sessions[0].encryption.cipher == ""
+    assert rsp.sessions[0].encryption.degree == "none"
+    assert rsp.sessions[0].signing
+    assert rsp.sessions[0].signing.cipher == "AES-128-GMAC"
+    assert rsp.sessions[0].signing.degree == "partial"
 
 
 def test_close_share(mock_grpc_server):