coercer.network.DCERPCSession

  1#!/usr/bin/env python3
  2# -*- coding: utf-8 -*-
  3# File name          : DCERPCSession.py
  4# Author             : Podalirius (@podalirius_)
  5# Date created       : 15 Sep 2022
  6
  7import sys
  8from impacket.dcerpc.v5 import transport
  9from impacket.uuid import uuidtup_to_bin
 10from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_WINNT, RPC_C_AUTHN_LEVEL_PKT_PRIVACY
 11
 12
 13class DCERPCSession(object):
 14    """
 15    Documentation for class DCERPCSession
 16    """
 17
 18    __rpctransport = None
 19    session = None
 20    target = None
 21
 22    def __init__(self, credentials, verbose=False):
 23        super(DCERPCSession, self).__init__()
 24        self.__verbose = True
 25        self.credentials = credentials
 26
 27    def connect_ncacn_np(self, target, pipe, targetIp=None, debug=False):
 28        """
 29
 30        """
 31        self.target = target
 32        ncan_target = r'ncacn_np:%s[%s]' % (target, pipe)
 33        self.__rpctransport = transport.DCERPCTransportFactory(ncan_target)
 34
 35        debug = False
 36
 37        if hasattr(self.__rpctransport, 'set_credentials'):
 38            self.__rpctransport.set_credentials(
 39                username=self.credentials.username,
 40                password=self.credentials.password,
 41                domain=self.credentials.domain,
 42                lmhash=self.credentials.lmhash,
 43                nthash=self.credentials.nthash
 44            )
 45
 46        if self.credentials.doKerberos == True:
 47            self.__rpctransport.set_kerberos(self.credentials.doKerberos, kdcHost=self.credentials.kdcHost)
 48        if targetIp is not None:
 49            self.__rpctransport.setRemoteHost(targetIp)
 50
 51        self.session = self.__rpctransport.get_dce_rpc()
 52        self.session.set_auth_type(RPC_C_AUTHN_WINNT)
 53        self.session.set_auth_level(RPC_C_AUTHN_LEVEL_PKT_PRIVACY)
 54
 55        # Connecting to named pipe
 56        if debug:
 57            print("   [>] Connecting to %s ... " % ncan_target, end="")
 58            sys.stdout.flush()
 59        try:
 60            self.session.connect()
 61        except Exception as e:
 62            if debug:
 63                print("\x1b[1;91mfail\x1b[0m")
 64                print("      [!] Something went wrong, check error status => %s" % str(e))
 65            return None
 66        else:
 67            if debug:
 68                print("\x1b[1;92msuccess\x1b[0m")
 69        return self.session
 70
 71    def bind(self, interface_uuid, interface_version, debug=False):
 72        """
 73
 74        """
 75        # Binding to interface
 76        if debug:
 77            print("   [>] Binding to interface <uuid='%s', version='%s'> ... " % (interface_uuid, interface_version), end="")
 78            sys.stdout.flush()
 79        try:
 80            self.session.bind(uuidtup_to_bin((interface_uuid, interface_version)))
 81        except Exception as e:
 82            if debug:
 83                print("\x1b[1;91mfail\x1b[0m")
 84                print("      [!] Something went wrong, check error status => %s" % str(e))
 85            return False
 86        else:
 87            if debug:
 88                print("\x1b[1;92msuccess\x1b[0m")
 89        return True
 90
 91    def set_verbose(self, value):
 92        """
 93        set_verbose(value)
 94
 95        Sets the current verbosity level
 96        """
 97        self.__verbose = value
 98
 99    def get_verbose(self):
100        """
101        get_verbose()
102
103        Gets the current verbosity level
104        """
105        return self.__verbose