Check live beacon URIs, certificate expiry dates and redirectors with sliver-py

sliver-redir.py

from rich.console import Console
from rich.tree import Tree
from rich.markdown import Markdown
import os, asyncio
import time
from datetime import datetime

import sliver, OpenSSL, ssl, urllib.parse
from datetime import datetime
console = Console()

CONFIG_DIR = os.path.join(os.path.expanduser("~"), ".sliver-client", "configs")

async def process_config(cfgfile):
    config = sliver.SliverClientConfig.parse_config_file(cfgfile)
    try:
        client = sliver.SliverClient(config)
        await client.connect()
    except:
        return
    version = await client.version()
    redir_data = {}
    beacons = await client.beacons()
    print_beacons = False
    for beacon in beacons:
        if beacon.NextCheckin > time.time():
            beacon_host = os.path.basename(cfgfile).split("_")[1].split(".c")[0]
            beacon_remotehost = beacon.RemoteAddress.split(':')[0]
            beacon_uri = beacon.ActiveC2

            if beacon_host not in redir_data.keys():
                redir_data[beacon_host] = {}
            if beacon_remotehost not in redir_data[beacon_host].keys():
                redir_data[beacon_host][beacon_remotehost] = set()

            beacon_data = beacon_uri

            if beacon_uri.startswith("https"):
                host, port = urllib.parse.urlparse(beacon_uri).hostname, urllib.parse.urlparse(beacon_uri).port
                if port is None:
                    port = 443
                cert = ssl.get_server_certificate((host, port))
                x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, cert)
                beacon_data += " exp. *" + str(datetime.strptime(x509.get_notAfter().decode('ascii'), '%Y%m%d%H%M%SZ'))+"*"


            redir_data[beacon_host][beacon_remotehost].add(beacon_data)
            

    sessions = await client.sessions()
    print_sessions = False
    for beacon in sessions:
        if not beacon.IsDead:
            beacon_host = os.path.basename(cfgfile).split("_")[1].split(".c")[0]
            beacon_remotehost = beacon.RemoteAddress.split(':')[0]
            beacon_uri = beacon.ActiveC2

            if beacon_host not in redir_data.keys():
                redir_data[beacon_host] = {}
            if beacon_remotehost not in redir_data[beacon_host].keys():
                redir_data[beacon_host][beacon_remotehost] = set()
            
            beacon_data = beacon_uri

            if beacon_uri.startswith("https"):
                host, port = urllib.parse.urlparse(beacon_uri).hostname, urllib.parse.urlparse(beacon_uri).port
                if port is None:
                    port = 443
                cert = ssl.get_server_certificate((host, port))
                x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, cert)
                beacon_data += " exp. *" + str(datetime.strptime(x509.get_notAfter().decode('ascii'), '%Y%m%d%H%M%SZ')) +"*"


            redir_data[beacon_host][beacon_remotehost].add(beacon_data)


    redir_tree = Tree("connection tree for "+str(os.path.basename(cfgfile)))
    for host in redir_data:
        for remotehost in redir_data[host]:
            for uri in redir_data[host][remotehost]:
                redir_tree.add("[bold]teamserver:[/bold] "+host).add("[bold]redirector:[/bold] "+remotehost).add("[bold]beacon URI:[/bold] "+uri)

    console.print(redir_tree)
    console.print(Markdown("---"))
    

async def main():
    tasks = [asyncio.create_task(process_config(os.path.join(CONFIG_DIR, cfg))) for cfg in os.listdir(CONFIG_DIR)]
    await asyncio.wait(tasks)

if __name__ == '__main__':
    loop = asyncio.new_event_loop()
    loop.run_until_complete(main())