add client and session tests

docs/conf.py

@@ -11,7 +11,7 @@
 # documentation root, use os.path.abspath to make it absolute, like shown here.
 import os
 import sys
-from subprocess import Popen, PIPE
+from subprocess import PIPE, Popen
 
 DOCS = os.path.dirname(os.path.abspath(__file__))
 sys.path.insert(0, os.path.abspath(os.path.join(DOCS, "..", "src")))

pyproject.toml

@@ -35,7 +35,8 @@ dependencies = [
 extra-dependencies = [
     "rich~=12.5",
     "black~=22.6", 
-    "isort~=5.10", 
+    "isort~=5.10",
+    "ward~=0.66.1b0" 
 ]
 
 [tool.hatch.envs.dev.scripts]
@@ -59,4 +60,4 @@ include = "^/src"
 extend-exclude = "^/src/silver/pb"
 
 [tool.isort]
-profile = "black"
+profile = "black"

scripts/protobufgen.py

@@ -1,10 +1,10 @@
 import os
-import pkg_resources
 from pathlib import Path
+
+import pkg_resources
 from grpc_tools import protoc
 from rich.console import Console
 
-
 console = Console(log_time=False, log_path=False)
 ROOT_DIR = Path(__file__).parents[1]
 os.chdir(ROOT_DIR)

src/sliver/client.py

@@ -16,7 +16,6 @@
 
 
 import logging
-from telnetlib import SE
 from typing import AsyncGenerator, Dict, Iterable, List, Optional, Union
 
 import grpc
@@ -235,8 +234,8 @@ class SliverClient(BaseClient):
         :type name: str
         :param timeout: gRPC timeout, defaults to 60 seconds
         :type timeout: int, optional
-        :return: Updated protobuf session object
-        :rtype: client_pb2.Session
+        :return: None
+        :rtype: None
         """
         rename_req = client_pb2.RenameReq(SessionID=session_id, Name=name)
         await self._stub.Rename(rename_req, timeout=timeout)
@@ -266,8 +265,23 @@ class SliverClient(BaseClient):
         )
         return list(beacons.Beacons)
 
-    async def rm_beacon(self, beacon_id: str, timeout=TIMEOUT) -> None:
-        """Remove a beacon
+    async def rename_beacon(self, beacon_id: str, name: str, timeout=TIMEOUT) -> None:
+        """Rename a beacon
+
+        :param beacon_id: Beacon ID to update
+        :type beacon_id: str
+        :param name: Rename beacon to this value
+        :type name: str
+        :param timeout: gRPC timeout, defaults to 60 seconds
+        :type timeout: int, optional
+        :return: None
+        :rtype: None
+        """
+        rename_req = client_pb2.RenameReq(BeaconID=beacon_id, Name=name)
+        await self._stub.Rename(rename_req, timeout=timeout)
+
+    async def kill_beacon(self, beacon_id: str, timeout=TIMEOUT) -> None:
+        """Kill a beacon
 
         :param beacon_id: Numeric beacon ID to remove
         :type beacon_id: str
@@ -283,7 +297,7 @@ class SliverClient(BaseClient):
         """Get a list of tasks for a beacon
 
         :param beacon_id: Beacon ID to get tasks for
-        :type beacon_id: sts
+        :type beacon_id: str
         :param timeout: gRPC timeout, defaults to 60 seconds
         :type timeout: int, optional
         :return: List of protobuf Task objects
@@ -322,6 +336,34 @@ class SliverClient(BaseClient):
         )
         return list(jobs.Active)
 
+    async def job_by_id(self, job_id: int, timeout=TIMEOUT) -> Optional[client_pb2.Job]:
+        """Get job by id
+
+        :param job_id: Beacon ID to get tasks for
+        :type job_id: str
+        :param timeout: gRPC timeout, defaults to 60 seconds
+        :type timeout: int, optional
+        :return: List of protobuf Job objects
+        :rtype: List[client_pb2.Job]
+        """
+        for job in await self.jobs(timeout=timeout):
+            if job.ID == job_id:
+                return job
+
+    async def job_by_port(
+        self, job_port: int, timeout=TIMEOUT
+    ) -> Optional[client_pb2.Job]:
+        """Get job by port
+
+        :param timeout: gRPC timeout, defaults to 60 seconds
+        :type timeout: int, optional
+        :return: List of protobuf Job objects
+        :rtype: List[client_pb2.Job]
+        """
+        for job in await self.jobs(timeout=timeout):
+            if job.Port == job_port:
+                return job
+
     async def kill_job(self, job_id: int, timeout=TIMEOUT) -> client_pb2.KillJob:
         """Kill a job
 
@@ -477,7 +519,7 @@ class SliverClient(BaseClient):
     async def start_https_listener(
         self,
         host: str = "0.0.0.0",
-        port: int = 80,
+        port: int = 443,
         website: str = "",
         domain: str = "",
         cert: bytes = b"",
@@ -627,7 +669,7 @@ class SliverClient(BaseClient):
         )
         return await self._stub.StartHTTPStagerListener(stage_req, timeout=timeout)
 
-    async def generate(
+    async def generate_implant(
         self, config: client_pb2.ImplantConfig, timeout: int = 360
     ) -> client_pb2.Generate:
         """Generate a new implant using a given configuration
@@ -642,7 +684,7 @@ class SliverClient(BaseClient):
         req = client_pb2.GenerateReq(Config=config)
         return await self._stub.Generate(req, timeout=timeout)
 
-    async def regenerate(
+    async def regenerate_implant(
         self, implant_name: str, timeout=TIMEOUT
     ) -> client_pb2.Generate:
         """Regenerate an implant binary given the implants "name"
@@ -708,7 +750,7 @@ class SliverClient(BaseClient):
             common_pb2.Empty(), timeout=timeout
         )
 
-    async def generate_unique_ip(self, timeout=TIMEOUT) -> client_pb2.UniqueWGIP:
+    async def generate_wg_ip(self, timeout=TIMEOUT) -> client_pb2.UniqueWGIP:
         """Generate a unique IP address for use with WireGuard
 
         :param timeout: gRPC timeout, defaults to 60 seconds
@@ -757,7 +799,7 @@ class SliverClient(BaseClient):
         """
         return await self._stub.SaveImplantProfile(profile, timeout=timeout)
 
-    async def msf_stage(
+    async def generate_msf_stager(
         self,
         arch: str,
         format: str,
@@ -801,7 +843,7 @@ class SliverClient(BaseClient):
         return await self._stub.MsfStage(stagerReq, timeout=timeout)
 
     async def shellcode(
-        self, data: bytes, function_name: str, arguments: str, timeout=TIMEOUT
+        self, data: bytes, function_name: str, arguments: str = "", timeout=TIMEOUT
     ) -> client_pb2.ShellcodeRDI:
         """Generate Donut shellcode
 
@@ -855,7 +897,7 @@ class SliverClient(BaseClient):
         :type timeout: int, optional
         """
         website = client_pb2.Website(Name=name)
-        await self._stub.Websites(website, timeout=timeout)
+        await self._stub.WebsiteRemove(website, timeout=timeout)
 
     async def add_website_content(
         self,
@@ -881,12 +923,10 @@ class SliverClient(BaseClient):
         :rtype: client_pb2.Website
         """
         web = client_pb2.WebContent(
-            Path=web_path, ContentType=content_type, Content=content
+            Path=web_path, ContentType=content_type, Content=content, Size=len(content)
         )
-        web.Size = len(content)
 
-        web_add = client_pb2.WebsiteAddContent(Name=name)
-        web_add.Contents[web_path] = web
+        web_add = client_pb2.WebsiteAddContent(Name=name, Contents={web_path: web})
         return await self._stub.WebsiteAddContent(web_add, timeout=timeout)
 
     async def update_website_content(
@@ -913,12 +953,10 @@ class SliverClient(BaseClient):
         :rtype: client_pb2.Website
         """
         web = client_pb2.WebContent(
-            Path=web_path, ContentType=content_type, Content=content
+            Path=web_path, ContentType=content_type, Content=content, Size=len(content)
         )
-        web.Size = len(content)
 
-        web_update = client_pb2.WebsiteAddContent(Name=name)
-        web_update.Contents[web_path] = web
+        web_update = client_pb2.WebsiteAddContent(Name=name, Contents={web_path: web})
         return await self._stub.WebsiteUpdateContent(web_update, timeout=timeout)
 
     async def remove_website_content(

src/sliver/interactive.py

@@ -16,6 +16,8 @@
 
 from typing import List, Literal, Optional
 
+from sliver.pb.commonpb import common_pb2
+
 from ._protocols import InteractiveObject
 from .protobuf import client_pb2, sliver_pb2
 
@@ -155,23 +157,25 @@ class BaseInteractiveCommands:
         return await self._stub.Mkdir(self._request(make), timeout=self.timeout)
 
     async def download(
-        self: InteractiveObject, remote_path: str
+        self: InteractiveObject, remote_path: str, recurse: bool = False
     ) -> sliver_pb2.Download:
-        """Download a file from the remote file system
+        """Download a file or directory from the remote file system
 
         :param remote_path: File to download
         :type remote_path: str
+        :param recurse: Download all files in a directory
+        :type recurse: bool
         :return: Protobuf Download object
         :rtype: sliver_pb2.Download
         """
-        download = sliver_pb2.DownloadReq(Path=remote_path)
+        download = sliver_pb2.DownloadReq(Path=remote_path, Recurse=recurse)
         return await self._stub.Download(self._request(download), timeout=self.timeout)
 
     async def upload(
         self: InteractiveObject,
         remote_path: str,
         data: bytes,
-        encoder: Literal["", "gzip"] = "",
+        is_ioc: bool = False,
     ) -> sliver_pb2.Upload:
         """Write data to specified path on remote file system
 
@@ -179,12 +183,12 @@ class BaseInteractiveCommands:
         :type remote_path: str
         :param data: Data to write
         :type data: bytes
-        :param encoder: Data encoder ('', 'gzip'), defaults to ''
-        :type encoder: str, optional
+        :param is_ioc: Data is an indicator of compromise, defaults to False
+        :type is_ioc: bool, optional
         :return: Protobuf Upload object
         :rtype: sliver_pb2.Upload
         """
-        upload = sliver_pb2.UploadReq(Path=remote_path, Data=data, Encoder=encoder)
+        upload = sliver_pb2.UploadReq(Path=remote_path, Data=data, IsIOC=is_ioc)
         return await self._stub.Upload(self._request(upload), timeout=self.timeout)
 
     async def process_dump(self: InteractiveObject, pid: int) -> sliver_pb2.ProcessDump:
@@ -532,7 +536,7 @@ class BaseInteractiveCommands:
         return await self._stub.GetEnv(self._request(env), timeout=self.timeout)
 
     async def set_env(
-        self: InteractiveObject, name: str, value: str
+        self: InteractiveObject, key: str, value: str
     ) -> sliver_pb2.SetEnv:
         """Set an environment variable
 
@@ -543,10 +547,20 @@ class BaseInteractiveCommands:
         :return: Protobuf SetEnv object
         :rtype: sliver_pb2.SetEnv
         """
-        env = sliver_pb2.SetEnvReq()
-        env.Variable.Key = name
-        env.Variable.Value = value
-        return await self._stub.SetEnv(self._request(env), timeout=self.timeout)
+        env_var = common_pb2.EnvVar(Key=key, Value=value)
+        env_req = sliver_pb2.SetEnvReq(Variable=env_var)
+        return await self._stub.SetEnv(self._request(env_req), timeout=self.timeout)
+
+    async def unset_env(self: InteractiveObject, key: str) -> sliver_pb2.UnsetEnv:
+        """Unset an environment variable
+
+        :param value: Value of the environment variable
+        :type value: str
+        :return: Protobuf SetEnv object
+        :rtype: sliver_pb2.SetEnv
+        """
+        env = sliver_pb2.UnsetEnvReq(Name=key)
+        return await self._stub.UnsetEnv(self._request(env), timeout=self.timeout)
 
     async def registry_read(
         self: InteractiveObject, hive: str, reg_path: str, key: str, hostname: str

tests/data/test_write.v

@@ -0,0 +1,13 @@
+import os
+
+[callconv: stdcall]
+[export: Main]
+fn output() {
+	mut output := os.open_file("test_write.txt", "w") or {return}
+	output.write_string("Hello, DLL!") or {return}
+	output.close()
+}
+
+fn main () {
+	output()
+}

tests/data/website.html

@@ -0,0 +1,8 @@
+<html>
+<head>
+<title>SliverPy Test</title>
+</head>
+<body>
+<h1>SliverPy Test</h1>
+</body>
+</html>

tests/data/website_update.html

@@ -0,0 +1,8 @@
+<html>
+<head>
+<title>SliverPy Test</title>
+</head>
+<body>
+<h1>SliverPy Test (UPDATED!)</h1>
+</body>
+</html>

tests/test_client.py

@@ -0,0 +1,306 @@
+import os
+from pathlib import Path
+
+from ward import fixture, test, skip
+
+from sliver import SliverClient, SliverClientConfig
+from sliver.pb.clientpb.client_pb2 import (
+    ImplantC2,
+    ImplantConfig,
+    ImplantProfile,
+    OutputFormat,
+    StageProtocol,
+)
+
+
+@fixture(scope="global")
+async def sliver_client() -> SliverClient:
+    CONFIG_PATH = Path("~/.sliver-client/configs/sliverpy.cfg").expanduser()
+    config = SliverClientConfig.parse_config_file(CONFIG_PATH)
+    client = SliverClient(config)
+    await client.connect()
+    return client
+
+
+@fixture(scope="global")
+async def implant_config() -> ImplantConfig:
+    return ImplantConfig(
+        IsBeacon=False,
+        Name="sliver-pytest-" + os.urandom(8).hex(),
+        GOARCH="amd64",
+        GOOS="linux",
+        Format=OutputFormat.EXECUTABLE,
+        ObfuscateSymbols=False,
+        C2=[ImplantC2(Priority=0, URL="http://localhost:80")],
+    )
+
+
+@fixture(scope="global")
+def sliverpy_random_name() -> str:
+    return "sliver-pytest-" + os.urandom(8).hex()
+
+
+@fixture(scope="global")
+def data_dir() -> Path:
+    return Path(__file__).parent / "data"
+
+
+@test("**Client can get version**")
+async def _(client: SliverClient = sliver_client):
+    assert await client.version()
+
+
+@test("Client can list operators")
+async def _(client: SliverClient = sliver_client):
+    assert await client.operators()
+
+
+@test("Client can list beacons")
+async def _(client: SliverClient = sliver_client):
+    assert await client.beacons()
+
+
+@test("Client can list beacons by ID")
+async def _(client: SliverClient = sliver_client):
+    beacons = await client.beacons()
+    assert await client.beacon_by_id(beacons[0].ID)
+
+
+@test("Client can rename a beacon")
+async def _(client: SliverClient = sliver_client):
+    beacons = await client.beacons()
+    beacon_name = beacons[0].Name
+    beacon_id = beacons[0].ID
+    await client.rename_beacon(beacon_id, "sliver-pytest")
+
+    beacon = await client.beacon_by_id(beacon_id)
+    assert beacon.Name == "sliver-pytest"
+
+    await client.rename_beacon(beacon.ID, beacon_name)
+
+
+@test("Client can list sessions")
+async def _(client: SliverClient = sliver_client):
+    assert await client.sessions()
+
+
+@test("Client can list sessions by ID")
+async def _(client: SliverClient = sliver_client):
+    sessions = await client.sessions()
+    assert await client.session_by_id(sessions[0].ID)
+
+
+@test("Client can rename a session")
+async def _(client: SliverClient = sliver_client):
+    sessions = await client.sessions()
+    session_name = sessions[0].Name
+    session_id = sessions[0].ID
+    await client.rename_session(session_id, "sliver-pytest")
+
+    session = await client.session_by_id(session_id)
+    assert session.Name == "sliver-pytest"
+
+    await client.rename_session(session.ID, session_name)
+
+
+@test("Client can list implant builds")
+async def _(client: SliverClient = sliver_client):
+    assert await client.implant_builds()
+
+
+@test("Client can generate a new implant")
+async def _(
+    client: SliverClient = sliver_client, config: ImplantConfig = implant_config
+):
+
+    assert await client.generate_implant(config)
+
+
+@test("Client can regenerate an implant")
+async def _(
+    client: SliverClient = sliver_client, config: ImplantConfig = implant_config
+):
+    assert await client.regenerate_implant(config.Name)
+
+
+@test("Client can save implant profiles")
+async def _(
+    client: SliverClient = sliver_client,
+    config: ImplantConfig = implant_config,
+    name: str = sliverpy_random_name,
+):
+    implant_profile = ImplantProfile(Name=name, Config=config)
+    assert await client.save_implant_profile(implant_profile)
+
+
+@test("Client can list implant profiles")
+async def _(client: SliverClient = sliver_client, name: str = sliverpy_random_name):
+    assert name in [profile.Name for profile in await client.implant_profiles()]
+
+
+@test("Client can delete implant profiles")
+async def _(client: SliverClient = sliver_client, name: str = sliverpy_random_name):
+    await client.delete_implant_profile(name)
+    assert name not in [profile.Name for profile in await client.implant_profiles()]
+
+
+@test("Client can delete implant builds")
+async def _(
+    client: SliverClient = sliver_client,
+    config: ImplantConfig = implant_config,
+):
+    await client.delete_implant_build(config.Name)
+    assert config.Name not in [build for build in await client.implant_builds()]
+
+
+@test("Client can list jobs")
+async def _(client: SliverClient = sliver_client):
+    assert await client.jobs()
+
+
+@test("Client can get job by ID")
+async def _(client: SliverClient = sliver_client):
+    jobs = await client.jobs()
+    assert await client.job_by_id(jobs[0].ID)
+
+
+@test("Client can get job by port")
+async def _(client: SliverClient = sliver_client):
+    assert await client.job_by_port(80)
+
+
+@test("Client can kill jobs")
+async def _(client: SliverClient = sliver_client):
+    jobs = await client.jobs()
+    for job in jobs:
+        if job.Port != 80:
+            await client.kill_job(job.ID)
+    assert len(await client.jobs()) == 1
+
+
+@test("Client can start HTTP listener on port 8080")
+async def _(client: SliverClient = sliver_client):
+    assert await client.start_http_listener()
+
+
+@test("Client can start HTTPS listener on port 8443")
+async def _(client: SliverClient = sliver_client):
+    assert await client.start_https_listener()
+
+
+@test("Client can start DNS listener on port 53")
+async def _(client: SliverClient = sliver_client):
+    assert await client.start_dns_listener(domains=["sliverpy.local"])
+
+
+@test("Client can start MTLS listener on port 8888")
+async def _(client: SliverClient = sliver_client):
+    assert await client.start_mtls_listener()
+
+
+@test("Client can start TCP stager listener on port 9000")
+async def _(client: SliverClient = sliver_client):
+    assert await client.start_tcp_stager_listener("0.0.0.0", 9000, b"sliver-pytest")
+
+
+@test("Client can start HTTP stager listener on port 9001")
+async def _(client: SliverClient = sliver_client):
+    assert await client.start_http_stager_listener("0.0.0.0", 9001, b"sliver-pytest")
+
+
+@skip("Cert generation not implemented")
+@test("Client can start HTTPS stager listener on port 9002")
+async def _(client: SliverClient = sliver_client):
+    assert await client.start_http_stager_listener("0.0.0.0", 9002, b"sliver-pytest")
+
+
+@test("Client can generate a WireGuard IP")
+async def _(client: SliverClient = sliver_client):
+    assert await client.generate_wg_ip()
+
+
+@skip("Something is wrong with killing WG listeners on the server")
+@test("Client can start WG listener on ports 5353/8889/1338")
+async def _(client: SliverClient = sliver_client):
+    ip = await client.generate_wg_ip()
+    print(ip.IP)
+    assert await client.start_wg_listener(ip.IP, 5353, 8889, 1338)
+
+
+@test("Client can generate a WireGuard client config")
+async def _(client: SliverClient = sliver_client):
+    assert await client.generate_wg_client_config()
+
+
+@test("Client can kill jobs (again) except WireGuard")
+async def _(client: SliverClient = sliver_client):
+    jobs = await client.jobs()
+    for job in jobs:
+        if job.Port != 80:
+            await client.kill_job(job.ID)
+    assert len(await client.jobs()) <= 2
+
+
+@test("Client can generate an MSF stager")
+async def _(client: SliverClient = sliver_client):
+    assert await client.generate_msf_stager(
+        arch="amd64",
+        format="raw",
+        host="127.0.0.1",
+        port=9000,
+        os="windows",
+        protocol=StageProtocol.TCP,
+        badchars=[],
+    )
+
+
+@test("Client can generate Donut shellcode")
+async def _(client: SliverClient = sliver_client, data_dir: Path = data_dir):
+    dll_data = Path(data_dir / "test_write.exe").read_bytes()
+    assert await client.shellcode(dll_data, "Main")
+
+
+@test("Client can interact with a session")
+async def _(client: SliverClient = sliver_client):
+    sessions = await client.sessions()
+    session = sessions[0]
+    assert await client.interact_session(session.ID)
+
+
+@test("Client can interact with a beacon")
+async def _(client: SliverClient = sliver_client):
+    beacons = await client.beacons()
+    beacon = beacons[0]
+    assert await client.interact_beacon(beacon.ID)
+
+
+@test("Client can add website content")
+async def _(client: SliverClient = sliver_client, data_dir: Path = data_dir):
+    html_content = Path(data_dir / "website.html").read_bytes()
+    assert await client.add_website_content(
+        "sliverpy-test", "sliverpy", "test/html", html_content
+    )
+
+
+@test("Client can update website content")
+async def _(client: SliverClient = sliver_client, data_dir: Path = data_dir):
+    html_content = Path(data_dir / "website_update.html").read_bytes()
+    assert await client.add_website_content(
+        "sliverpy-test", "sliverpy", "test/html", html_content
+    )
+
+
+@test("Client can list websites")
+async def _(client: SliverClient = sliver_client):
+    assert "sliverpy-test" in [website.Name for website in await client.websites()]
+
+
+@test("Client can remove website content")
+async def _(client: SliverClient = sliver_client, data_dir: Path = data_dir):
+    assert await client.remove_website_content("sliverpy-test", ["sliverpy"])
+
+
+@test("Client can remove website")
+async def _(client: SliverClient = sliver_client, data_dir: Path = data_dir):
+    await client.remove_website("sliverpy-test")
+    assert "sliverpy-test" not in [website.Name for website in await client.websites()]

tests/test_interactive.py

@@ -0,0 +1,112 @@
+from pathlib import Path
+from ward import test, fixture
+from sliver import SliverClient
+from sliver.session import InteractiveSession
+
+from .test_client import data_dir, sliver_client, sliverpy_random_name
+
+
+@fixture(scope="module")
+async def session_zero(client: SliverClient = sliver_client) -> InteractiveSession:
+    sessions = await client.sessions()
+    return await client.interact_session(sessions[0].ID)  # type: ignore
+
+
+@test("InteractiveObject can send ping to server", tags=["interactive"])
+async def _(session: InteractiveSession = session_zero):
+    assert await session.ping()
+
+
+@test("InteractiveObject can list processes", tags=["interactive"])
+async def _(session: InteractiveSession = session_zero):
+    assert await session.ps()
+
+
+@test("InteractiveObject can get network interfaces", tags=["interactive"])
+async def _(session: InteractiveSession = session_zero):
+    assert await session.ifconfig()
+
+
+@test("InteractiveObject can get network connections", tags=["interactive"])
+async def _(session: InteractiveSession = session_zero):
+    assert await session.netstat(True, True, True, True, True)
+
+
+@test("InteractiveObject can get working directory", tags=["interactive"])
+async def _(session: InteractiveSession = session_zero):
+    assert await session.pwd()
+
+
+@test("InteractiveObject can list directory", tags=["interactive"])
+async def _(session: InteractiveSession = session_zero):
+    assert await session.ls()
+
+
+@test("InteractiveObject can change directory", tags=["interactive"])
+async def _(session: InteractiveSession = session_zero):
+    assert await session.cd(".")
+
+
+@test("InteractiveObject can make a directory", tags=["interactive"])
+async def _(
+    session: InteractiveSession = session_zero, target_dir: str = sliverpy_random_name
+):
+    assert await session.mkdir(target_dir)
+
+
+@test("InteractiveObject can upload a file", tags=["interactive"])
+async def _(
+    session: InteractiveSession = session_zero,
+    target_dir: str = sliverpy_random_name,
+):
+    assert await session.upload(target_dir + "/sliverpy.txt", b"sliverpy")
+
+
+@test("InteractiveObject can download files", tags=["interactive"])
+async def _(
+    session: InteractiveSession = session_zero,
+    target_dir: str = sliverpy_random_name,
+):
+    assert await session.download(target_dir, True)
+
+
+@test("InteractiveObject can remove a directory", tags=["interactive"])
+async def _(
+    session: InteractiveSession = session_zero, path: str = sliverpy_random_name
+):
+    assert await session.rm(path, recursive=True, force=True)
+
+
+@test("InteractiveObject can set an environment variable", tags=["interactive"])
+async def _(
+    session: InteractiveSession = session_zero, value: str = sliverpy_random_name
+):
+    assert await session.set_env("SLIVERPY_TEST", value)
+
+
+@test("InteractiveObject can get an environment variable", tags=["interactive"])
+async def _(
+    session: InteractiveSession = session_zero, value: str = sliverpy_random_name
+):
+    assert await session.get_env(value)
+
+
+@test("InteractiveObject can unset an environment variable", tags=["interactive"])
+async def _(
+    session: InteractiveSession = session_zero, value: str = sliverpy_random_name
+):
+    assert await session.unset_env(value)
+
+
+@test("InteractiveObject can take a screenshot", tags=["interactive"])
+async def _(
+    session: InteractiveSession = session_zero, value: str = sliverpy_random_name
+):
+    assert await session.screenshot()
+
+
+@test("InteractiveObject can take a memory dump", tags=["interactive"])
+async def _(
+    session: InteractiveSession = session_zero, value: str = sliverpy_random_name
+):
+    assert await session.process_dump(session.pid)