modified inits for classes

docs/api/configuration.rst

@@ -3,6 +3,5 @@ Configuration
 
 
 .. autoclass:: sliver.config.SliverClientConfig
-    :no-special-members:
     :members:
     :undoc-members:

docs/conf.py

@@ -39,7 +39,6 @@
 
 autodoc_default_options = {
     "members": True,
-    "member-order": "bysource",
     "special-members": False,
     "undoc-members": True,
     "exclude-members": "__weakref__, DESCRIPTOR",

src/sliver/beacon.py

@@ -17,7 +17,7 @@
 import asyncio
 import functools
 import logging
-from typing import Union
+from typing import Any, Dict, Tuple, Union
 
 import grpc
 
@@ -29,133 +29,155 @@
 
 
 class BaseBeacon(object):
-
-    _beacon: client_pb2.Beacon
-    beacon_tasks = {}
-
     def __init__(
         self,
         beacon: client_pb2.Beacon,
         channel: grpc.Channel,
         timeout: int = TIMEOUT,
-        logger: Union[logging.Handler, None] = None,
     ):
+        """Base class for Beacon classes.
+
+        :param beacon: Beacon protobuf object.
+        :type beacon: client_pb2.Beacon
+        :param channel: A gRPC channel.
+        :type channel: grpc.Channel
+        :param timeout: Seconds to wait for timeout, defaults to TIMEOUT
+        :type timeout: int, optional
+        """
         self._log = logging.getLogger(self.__class__.__name__)
-        self._channel = channel
-        self._beacon = beacon
+        self._channel: grpc.Channel = channel
+        self._beacon: client_pb2.Beacon = beacon
         self._stub = SliverRPCStub(channel)
         self.timeout = timeout
+        self.beacon_tasks: Dict[str, Tuple[asyncio.Future, Any]] = {}
         asyncio.get_event_loop().create_task(self.taskresult_events())
 
-    def _request(self, pb):
-        """
-        Set request attributes based on current beacon, I'd prefer to return a generic Request
-        object, but protobuf for whatever reason doesn't let you assign this type of field directly.
-
-        `pb` in this case is any protobuf message with a .Request field.
-
-        :param pb: A protobuf request object.
-        """
-        pb.Request.SessionID = self._beacon.ID
-        pb.Request.Timeout = self.timeout - 1
-        pb.Request.Async = True
-        return pb
-
-    async def taskresult_events(self):
-        """
-        Monitor task events for results, resolve futures for any results
-        we get back.
-        """
-        async for event in self._stub.Events(common_pb2.Empty()):
-            if event.EventType != "beacon-taskresult":
-                continue
-            try:
-                beacon_task = client_pb2.BeaconTask()
-                beacon_task.ParseFromString(event.Data)
-                if beacon_task.ID not in self.beacon_tasks:
-                    continue
-                task_content = await self._stub.GetBeaconTaskContent(
-                    client_pb2.BeaconTask(ID=beacon_task.ID)
-                )
-                task_future, pb_object = self.beacon_tasks[beacon_task.ID]
-                del self.beacon_tasks[beacon_task.ID]
-                if pb_object is not None:
-                    result = pb_object()
-                    result.ParseFromString(task_content.Response)
-                else:
-                    result = None
-                task_future.set_result(result)
-            except Exception as err:
-                self._log.exception(err)
-
     @property
     def beacon_id(self) -> int:
+        """Beacon ID"""
         return self._beacon.ID
 
     @property
     def name(self) -> str:
+        """Beacon name"""
         return self._beacon.Name
 
     @property
     def hostname(self) -> int:
+        """Beacon hostname"""
         return self._beacon.Hostname
 
     @property
     def uuid(self) -> str:
+        """Beacon UUID"""
         return self._beacon.UUID
 
     @property
     def username(self) -> str:
+        """Username"""
         return self._beacon.Username
 
     @property
     def uid(self) -> str:
+        """User ID"""
         return self._beacon.UID
 
     @property
     def gid(self) -> str:
+        """Group ID"""
         return self._beacon.GID
 
     @property
     def os(self) -> str:
+        """Operating system"""
         return self._beacon.OS
 
     @property
     def arch(self) -> str:
+        """Architecture"""
         return self._beacon.Arch
 
     @property
     def transport(self) -> str:
+        """Transport Method"""
         return self._beacon.Transport
 
     @property
     def remote_address(self) -> str:
+        """Remote address"""
         return self._beacon.RemoteAddress
 
     @property
     def pid(self) -> int:
+        """Process ID"""
         return self._beacon.PID
 
     @property
     def filename(self) -> str:
+        """Beacon filename"""
         return self._beacon.Filename
 
     @property
     def last_checkin(self) -> str:
+        """Last check in time"""
         return self._beacon.LastCheckin
 
     @property
     def active_c2(self) -> str:
+        """Active C2"""
         return self._beacon.ActiveC2
 
     @property
     def version(self) -> str:
+        """Version"""
         return self._beacon.Version
 
     @property
     def reconnect_interval(self) -> int:
+        """Reconnect interval"""
         return self._beacon.ReconnectInterval
 
+    def _request(self, pb):
+        """
+        Set request attributes based on current beacon, I'd prefer to return a generic Request
+        object, but protobuf for whatever reason doesn't let you assign this type of field directly.
+
+        `pb` in this case is any protobuf message with a .Request field.
+
+        :param pb: A protobuf request object.
+        """
+        pb.Request.SessionID = self._beacon.ID
+        pb.Request.Timeout = self.timeout - 1
+        pb.Request.Async = True
+        return pb
+
+    async def taskresult_events(self):
+        """
+        Monitor task events for results, resolve futures for any results
+        we get back.
+        """
+        async for event in self._stub.Events(common_pb2.Empty()):
+            if event.EventType != "beacon-taskresult":
+                continue
+            try:
+                beacon_task = client_pb2.BeaconTask()
+                beacon_task.ParseFromString(event.Data)
+                if beacon_task.ID not in self.beacon_tasks:
+                    continue
+                task_content = await self._stub.GetBeaconTaskContent(
+                    client_pb2.BeaconTask(ID=beacon_task.ID)
+                )
+                task_future, pb_object = self.beacon_tasks[beacon_task.ID]
+                del self.beacon_tasks[beacon_task.ID]
+                if pb_object is not None:
+                    result = pb_object()
+                    result.ParseFromString(task_content.Response)
+                else:
+                    result = None
+                task_future.set_result(result)
+            except Exception as err:
+                self._log.exception(err)
+
 
 def beacon_taskresult(pb_object):
     """

src/sliver/config.py

@@ -13,14 +13,29 @@
     You should have received a copy of the GNU General Public License
     along with this program.  If not, see <https://www.gnu.org/licenses/>.
 """
+from __future__ import annotations
 
 import json
-from typing import Type, TypeVar, Union
-
-T = TypeVar("T", bound="SliverClientConfig")
+from typing import Type, Union
 
 
 class SliverClientConfig(object):
+    """
+    This class parses and represents Sliver operator configuration files, typically this class is automatically
+    instantiated using one of the class methods :class:`SliverClientConfig.parse_config()` or :class:`SliverClientConfig.parse_config_file()` but can be directly
+    instantiated too.
+
+    :param operator: Operator name, note that this value is only used by the client and is ignored by the server.
+    :param lhost: The listener host to connect to (i.e., the Sliver server host).
+    :param lhost: The TCP port of the host listener (i.e., the TCP port of the Sliver "multiplayer" service).
+    :param ca_certificate: The Sliver server certificate authority.
+    :param certificate: The mTLS client certificate.
+    :param private_key: The mTLS private key.
+    :param token: The user's authentication token.
+
+    :raises ValueError: A parameter contained an invalid value.
+    """
+
     def __init__(
         self,
         operator: str,
@@ -31,21 +46,6 @@ def __init__(
         private_key: str,
         token: str,
     ):
-        """
-        This class parses and represents Sliver operator configuration files, typically this class is automatically
-        instantiated using one of the class methods :class:`SliverClientConfig.parse_config()` or :class:`SliverClientConfig.parse_config_file()` but can be directly
-        instantiated too.
-
-        :param operator: Operator name, note that this value is only used by the client and is ignored by the server.
-        :param lhost: The listener host to connect to (i.e., the Sliver server host).
-        :param lhost: The TCP port of the host listener (i.e., the TCP port of the Sliver "multiplayer" service).
-        :param ca_certificate: The Sliver server certificate authority.
-        :param certificate: The mTLS client certificate.
-        :param private_key: The mTLS private key.
-        :param token: The user's authentication token.
-
-        :raises ValueError: A parameter contained an invalid value.
-        """
         self.operator = operator
         self.lhost = lhost
         if not 0 < lport < 65535:
@@ -73,25 +73,25 @@ def __repr__(self):
         )
 
     @classmethod
-    def parse_config(cls: Type[T], data: Union[str, bytes]) -> T:
+    def parse_config(cls, data: Union[str, bytes]) -> SliverClientConfig:
         """Parses the content of a Sliver operator configuration file and
         returns the instantiated :class:`SliverClientConfig`
 
         :param data: The Sliver operator configuration file content.
         :type data: Union[str, bytes]
         :return: An instantiated :class:`SliverClientConfig` object.
-        :rtype: T
+        :rtype: SliverClientConfig
         """
         return cls(**json.loads(data))
 
     @classmethod
-    def parse_config_file(cls: Type[T], filepath: str) -> T:
+    def parse_config_file(cls, filepath: str) -> SliverClientConfig:
         """Parse a given file path as a Sliver operator configuration file.
 
         :param filepath: File system path to an operator configuration file.
         :type filepath: str
         :return: An instantiated :class:`SliverClientConfig` object.
-        :rtype: T
+        :rtype: SliverClientConfig
         """
         with open(filepath, "r") as fp:
             data = fp.read()