aniqfakhrul · aniqfakhrul · May 17, 2025 · May 17, 2025 · May 17, 2025 · May 17, 2025
diff --git a/powerview/_version.py b/powerview/_version.py
@@ -1,7 +1,7 @@
 import datetime
 
 __year__ = datetime.date.today().year
-__version__ = f"{__year__}.0.8"
+__version__ = f"{__year__}.0.9"
 __author__ = [
 	"Aniq Fakhrul",
 	"Ali Radzali"

diff --git a/powerview/modules/smbclient.py b/powerview/modules/smbclient.py
@@ -16,40 +16,137 @@ class SMBClient:
     def __init__(self, client):
         self.client = client
 
+    def _parse_dacl(self, secDesc):
+        sd = ldaptypes.SR_SECURITY_DESCRIPTOR()
+        if isinstance(secDesc, list):
+            secDesc = b''.join(secDesc)
+        sd.fromString(secDesc)
+
+        security_info = {}
+        if sd['OwnerSid'] is not None:
+            security_info['OwnerSid'] = sd['OwnerSid'].formatCanonical()
+        if sd['GroupSid'] is not None:
+            security_info['GroupSid'] = sd['GroupSid'].formatCanonical()
+
+        try:
+            dacl_data = sd['Dacl']['Data']
+            aces = []
+            for ace_obj in dacl_data:
+                ace_type_name = ace_obj['TypeName']
+
+                # Filter for ACE types we can parse well
+                if ace_type_name not in ["ACCESS_ALLOWED_ACE", "ACCESS_DENIED_ACE", 
+                                            "ACCESS_ALLOWED_OBJECT_ACE", "ACCESS_DENIED_OBJECT_ACE",
+                                            "SYSTEM_AUDIT_ACE", "SYSTEM_ALARM_ACE",
+                                            "SYSTEM_AUDIT_OBJECT_ACE", "SYSTEM_ALARM_OBJECT_ACE"]:
+                    logging.debug(f"[SMBClient: get_file_info] Skipping unhandled ACE type: {ace_type_name}")
+                    continue
+
+                trustee_sid_str = ace_obj['Ace']['Sid'].formatCanonical()
+
+                ace_flags_int = ace_obj['AceFlags']
+                parsed_ace_flags_list = [FLAG.name for FLAG in ACE_FLAGS if ace_flags_int & FLAG.value]
+
+                access_mask_int = ace_obj['Ace']['Mask']['Mask']
+                parsed_permissions_list = self._parsePerms(access_mask_int)
+
+                if not parsed_permissions_list and access_mask_int != 0: # If no known flags matched but mask is not zero
+                    parsed_permissions_list.append(f"UNKNOWN_MASK_0x{access_mask_int:08X}")
+
+                # Initialize object-specific fields
+                parsed_object_ace_specific_flags_list = None
+                obj_type_guid_str = None
+                inh_obj_type_guid_str = None
+
+                if ace_type_name in ["ACCESS_ALLOWED_OBJECT_ACE", "ACCESS_DENIED_OBJECT_ACE", "SYSTEM_AUDIT_OBJECT_ACE", "SYSTEM_ALARM_OBJECT_ACE"]:
+                    object_ace_specific_flags_int = ace_obj['Ace']['Flags']
+                    parsed_object_ace_specific_flags_list = [FLAG.name for FLAG in OBJECT_ACE_FLAGS if object_ace_specific_flags_int & FLAG.value]
+
+                    if ace_obj['Ace']['ObjectTypeLen'] != 0:
+                        obj_type_guid_str = bin_to_string(ace_obj['Ace']['ObjectType']).lower()
+
+                    if ace_obj['Ace']['InheritedObjectTypeLen'] != 0:
+                        inh_obj_type_guid_str = bin_to_string(ace_obj['Ace']['InheritedObjectType']).lower()
+
+                ace_info_entry = {
+                    'type': ace_type_name,
+                    'trustee': trustee_sid_str,
+                    'ace_flags': parsed_ace_flags_list,
+                    'access_mask_raw': access_mask_int,
+                    'permissions': parsed_permissions_list,
+                    'object_ace_specific_flags': parsed_object_ace_specific_flags_list,
+                    'object_type_guid': obj_type_guid_str,
+                    'inherited_object_type_guid': inh_obj_type_guid_str
+                }
+                aces.append(ace_info_entry)
+
+            security_info['Dacl'] = aces
+            return security_info
+        except Exception as e:
+            logging.error(f"[SMBClient: get_file_info] Error parsing security descriptor: {e}")
+            import traceback
+            logging.debug(f"[SMBClient: get_file_info] Traceback: {traceback.format_exc()}")
+            # Store partial info if available
+            if 'security_info' not in locals(): security_info = {} # ensure security_info exists
+            if 'Dacl' not in security_info : security_info['Dacl'] = [] # ensure Dacl list exists
+            return security_info # Assign even if parsing failed mid-way
+
     def shares(self):
         if self.client is None:
             logging.error("[SMBClient: shares] Not logged in")
             return
+
         return self.client.listShares()
 
     def share_info(self, share):
-        """
-        Get detailed information about a share.
-
-        shi503_netname:                  'C$\x00' 
-        shi503_type:                     2147483648 
-        shi503_remark:                   'Default share\x00' 
-        shi503_permissions:              0 
-        shi503_max_uses:                 4294967295 
-        shi503_current_uses:             0 
-        shi503_path:                     'C:\\\x00' 
-        shi503_passwd:                   NULL 
-        shi503_servername:               '*\x00' 
-        shi503_reserved:                 0 
-        shi503_security_descriptor:      NULL None
-        """
         if self.client is None:
             logging.error("[SMBClient: share_info] Not logged in")
             return
 
-        from impacket.dcerpc.v5 import transport, srvs
         rpctransport = transport.SMBTransport(self.client.getRemoteName(), self.client.getRemoteHost(), filename=r'\srvsvc',
                                               smb_connection=self.client)
         dce = rpctransport.get_dce_rpc()
         dce.connect()
         dce.bind(srvs.MSRPC_UUID_SRVS)
-        resp = srvs.hNetrShareGetInfo(dce, share + '\x00', 503)
-        return resp['InfoStruct']['ShareInfo503']
+
+        share_info = {
+            'name': None,
+            'type': None,
+            'remark': None,
+            'path': None,
+            'permissions': None,
+            'max_uses': None,
+            'current_uses': None,
+            'passwd': None,
+            'servername': None,
+            'reserved': None,
+            'sd_info': None
+        }
+
+        try:
+            base_info = srvs.hNetrShareGetInfo(dce, share + '\x00', 1)
+            share_info['name'] = base_info['InfoStruct']['ShareInfo1']['shi1_netname'][:-1]
+            share_info['type'] = base_info['InfoStruct']['ShareInfo1']['shi1_type']
+            share_info['remark'] = base_info['InfoStruct']['ShareInfo1']['shi1_remark'][:-1]
+        except Exception as e:
+            logging.error(f"[SMBClient: share_info] Error getting share info via NetrShareGetInfo(Level 1): {e}")
+
+        try:
+            resp = srvs.hNetrShareGetInfo(dce, share + '\x00', 503)
+            share_info['path'] = resp['InfoStruct']['ShareInfo503']['shi503_path'][:-1]
+            share_info['permissions'] = resp['InfoStruct']['ShareInfo503']['shi503_permissions']
+            share_info['max_uses'] = resp['InfoStruct']['ShareInfo503']['shi503_max_uses']
+            share_info['current_uses'] = resp['InfoStruct']['ShareInfo503']['shi503_current_uses']
+            share_info['passwd'] = resp['InfoStruct']['ShareInfo503']['shi503_passwd']
+            share_info['servername'] = resp['InfoStruct']['ShareInfo503']['shi503_servername'][:-1]
+            share_info['reserved'] = resp['InfoStruct']['ShareInfo503']['shi503_reserved']
+            secDesc = resp['InfoStruct']['ShareInfo503']['shi503_security_descriptor']
+            if secDesc and len(secDesc) > 0:
+                share_info['sd_info'] = self._parse_dacl(secDesc)
+        except Exception as e:
+            logging.error(f"[SMBClient: share_info] Error getting share info via NetrShareGetInfo(Level 503): {e}")
+
+        return share_info
 
     def ls(self, share, path=''):
         if self.client is None:
@@ -265,79 +362,10 @@ def get_file_info(self, share, path):
                     path = path.replace('/', '\\')
 
                     resp = srvs.hNetrpGetFileSecurity(dce, share+'\x00', path+'\x00', security_flags)
-                    sd = ldaptypes.SR_SECURITY_DESCRIPTOR()
-                    sd.fromString(resp)
-
-                    security_info = {}
-                    if sd['OwnerSid'] is not None:
-                        security_info['OwnerSid'] = sd['OwnerSid'].formatCanonical()
-                    if sd['GroupSid'] is not None:
-                        security_info['GroupSid'] = sd['GroupSid'].formatCanonical()
-
-                    try:
-                        dacl_data = sd['Dacl']['Data']
-                        aces = []
-                        for ace_obj in dacl_data:
-                            ace_type_name = ace_obj['TypeName']
-
-                            # Filter for ACE types we can parse well
-                            if ace_type_name not in ["ACCESS_ALLOWED_ACE", "ACCESS_DENIED_ACE", 
-                                                     "ACCESS_ALLOWED_OBJECT_ACE", "ACCESS_DENIED_OBJECT_ACE",
-                                                     "SYSTEM_AUDIT_ACE", "SYSTEM_ALARM_ACE",
-                                                     "SYSTEM_AUDIT_OBJECT_ACE", "SYSTEM_ALARM_OBJECT_ACE"]:
-                                logging.debug(f"[SMBClient: get_file_info] Skipping unhandled ACE type: {ace_type_name}")
-                                continue
-
-                            trustee_sid_str = ace_obj['Ace']['Sid'].formatCanonical()
-
-                            ace_flags_int = ace_obj['AceFlags']
-                            parsed_ace_flags_list = [FLAG.name for FLAG in ACE_FLAGS if ace_flags_int & FLAG.value]
-
-                            access_mask_int = ace_obj['Ace']['Mask']['Mask']
-                            parsed_permissions_list = self._parsePerms(access_mask_int)
-
-                            if not parsed_permissions_list and access_mask_int != 0: # If no known flags matched but mask is not zero
-                                parsed_permissions_list.append(f"UNKNOWN_MASK_0x{access_mask_int:08X}")
-
-                            # Initialize object-specific fields
-                            parsed_object_ace_specific_flags_list = None
-                            obj_type_guid_str = None
-                            inh_obj_type_guid_str = None
-
-                            if ace_type_name in ["ACCESS_ALLOWED_OBJECT_ACE", "ACCESS_DENIED_OBJECT_ACE", "SYSTEM_AUDIT_OBJECT_ACE", "SYSTEM_ALARM_OBJECT_ACE"]:
-                                object_ace_specific_flags_int = ace_obj['Ace']['Flags']
-                                parsed_object_ace_specific_flags_list = [FLAG.name for FLAG in OBJECT_ACE_FLAGS if object_ace_specific_flags_int & FLAG.value]
-
-                                if ace_obj['Ace']['ObjectTypeLen'] != 0:
-                                    obj_type_guid_str = bin_to_string(ace_obj['Ace']['ObjectType']).lower()
-
-                                if ace_obj['Ace']['InheritedObjectTypeLen'] != 0:
-                                    inh_obj_type_guid_str = bin_to_string(ace_obj['Ace']['InheritedObjectType']).lower()
-
-                            ace_info_entry = {
-                                'type': ace_type_name,
-                                'trustee': trustee_sid_str,
-                                'ace_flags': parsed_ace_flags_list,
-                                'access_mask_raw': access_mask_int,
-                                'permissions': parsed_permissions_list,
-                                'object_ace_specific_flags': parsed_object_ace_specific_flags_list,
-                                'object_type_guid': obj_type_guid_str,
-                                'inherited_object_type_guid': inh_obj_type_guid_str
-                            }
-                            aces.append(ace_info_entry)
-
-                        security_info['Dacl'] = aces
-                        info['sd_info'] = security_info
-                    except Exception as e:
-                        logging.error(f"[SMBClient: get_file_info] Error parsing security descriptor: {e}")
-                        import traceback
-                        logging.debug(f"[SMBClient: get_file_info] Traceback: {traceback.format_exc()}")
-                        # Store partial info if available
-                        if 'security_info' not in locals(): security_info = {} # ensure security_info exists
-                        if 'Dacl' not in security_info : security_info['Dacl'] = [] # ensure Dacl list exists
-                        info['sd_info'] = security_info # Assign even if parsing failed mid-way
+                    if resp and len(resp) > 0:
+                        info['sd_info'] = self._parse_dacl(resp)
                 except Exception as rpc_error:
-                    logging.debug(f"[SMBClient: get_file_info] RPC error: {rpc_error}")
+                    raise Exception(f"[SMBClient: get_file_info] RPC error: {rpc_error}")
             return info
 
         except Exception as e:

diff --git a/powerview/powerview.py b/powerview/powerview.py
@@ -153,6 +153,7 @@ def _initialize_attributes_from_connection(self):
 		self.forest_dn = self.ldap_server.info.other["rootDomainNamingContext"][0] if isinstance(self.ldap_server.info.other["rootDomainNamingContext"], list) else self.ldap_server.info.other["rootDomainNamingContext"]
 		self.root_dn = self.ldap_server.info.other["defaultNamingContext"][0] if isinstance(self.ldap_server.info.other["defaultNamingContext"], list) else self.ldap_server.info.other["defaultNamingContext"]
 		self.configuration_dn = self.ldap_server.info.other["configurationNamingContext"][0] if isinstance(self.ldap_server.info.other["configurationNamingContext"], list) else self.ldap_server.info.other["configurationNamingContext"]
+		self.schema_dn = self.ldap_server.info.other["schemaNamingContext"][0] if isinstance(self.ldap_server.info.other["schemaNamingContext"], list) else self.ldap_server.info.other["schemaNamingContext"]
 		if not self.domain:
 			self.domain = dn2domain(self.root_dn)
 		self.flatName = self.ldap_server.info.other["ldapServiceName"][0].split("@")[-1].split(".")[0] if isinstance(self.ldap_server.info.other["ldapServiceName"], list) else self.ldap_server.info.other["ldapServiceName"].split("@")[-1].split(".")[0]
@@ -2265,7 +2266,7 @@ def add_domainou(self, identity, basedn=None, args=None):
 
 
 		ou_data = {
-				'objectCategory': 'CN=Organizational-Unit,CN=Schema,CN=Configuration,%s' % self.root_dn,
+				'objectCategory': f'CN=Organizational-Unit,{self.schema_dn}',
 				'name': identity,
 				}
 
@@ -3243,7 +3244,7 @@ def add_domaingroup(self, groupname, basedn=None, args=None):
 		ucd = {
 			'displayName': groupname,
 			'sAMAccountName': groupname,
-			'objectCategory': 'CN=Group,CN=Schema,CN=Configuration,%s' % self.root_dn,
+			'objectCategory': f'CN=Group,{self.schema_dn}',
 			'objectClass': ['top', 'group'],
 		}
 
@@ -3566,7 +3567,7 @@ def add_domaindnsrecord(self, recordname, recordaddress, zonename=None):
 		DNS_UTIL.get_next_serial(self.dc_ip, zonename, True)
 		node_data = {
 				# Schema is in the root domain (take if from schemaNamingContext to be sure)
-				'objectCategory': 'CN=Dns-Node,CN=Schema,CN=Configuration,%s' % self.root_dn,
+				'objectCategory': f'CN=Dns-Node,{self.schema_dn}',
 				'dNSTombstoned': "FALSE", # Need to hardcoded because of Kerberos issue, will revisit.
 				'name': recordname
 				}

diff --git a/powerview/utils/connections.py b/powerview/utils/connections.py
@@ -1142,12 +1142,19 @@ def ldap3_kerberos_login(self, connection, target, user, password, domain='', lm
 
 		return True
 
-	def init_smb_session(self, host, username=None, password=None, nthash=None, lmhash=None, domain=None, timeout=10, useCache=True):
+	def init_smb_session(self, host, username=None, password=None, nthash=None, lmhash=None, aesKey=None, domain=None, timeout=10, useCache=True):
 		try:
 			username = username or self.username
 			password = password or self.password 
 			nthash = nthash or self.nthash
 			lmhash = lmhash or self.lmhash
+			aesKey = aesKey or self.auth_aes_key
+			if aesKey:
+				useKerberos = True
+				useCache = False
+			else:
+				useKerberos = self.use_kerberos
+
 			domain = domain or self.domain
 
 			if host in self.smb_sessions:
@@ -1156,12 +1163,11 @@ def init_smb_session(self, host, username=None, password=None, nthash=None, lmha
 
 			logging.debug(f"[Connection: init_smb_session] Default timeout is set to {timeout}. Expect a delay")
 			conn = SMBConnection(host, host, sess_port=445, timeout=timeout)
-			if self.use_kerberos:
+			if useKerberos:
 				if self.TGT and self.TGS:
 					useCache = False
 
 				if useCache:
-					# only import if used
 					import os
 					from impacket.krb5.ccache import CCache
 					from impacket.krb5.kerberosv5 import KerberosError
@@ -1182,7 +1188,6 @@ def init_smb_session(self, host, username=None, password=None, nthash=None, lmha
 
 						creds = ccache.getCredential(principal)
 						if creds is None:
-							# Let's try for the TGT and go from there
 							principal = 'krbtgt/%s@%s' % (domain.upper(), domain.upper())
 							creds = ccache.getCredential(principal)
 							if creds is not None:
@@ -1194,28 +1199,27 @@ def init_smb_session(self, host, username=None, password=None, nthash=None, lmha
 							self.TGS = creds.toTGS(principal)
 							logging.debug('Using TGS from cache')
 
-						# retrieve user information from CCache file if needed
 						if username == '' and creds is not None:
 							username = creds['client'].prettyPrint().split(b'@')[0].decode('utf-8')
 							logging.debug('Username retrieved from CCache: %s' % username)
 						elif username == '' and len(ccache.principal.components) > 0:
 							username = ccache.principal.components[0]['data'].decode('utf-8')
 							logging.debug('Username retrieved from CCache: %s' % username)
 
-				conn.kerberosLogin(username, password, domain, lmhash, nthash, self.auth_aes_key, self.dc_ip, self.TGT, self.TGS)
+				conn.kerberosLogin(username, password, domain, lmhash, nthash, aesKey, self.dc_ip, self.TGT, self.TGS)
 			else:
 				conn.login(username, password, domain, lmhash, nthash)
 			self.smb_sessions[host] = conn
 			return conn
 		except OSError as e:
 			logging.debug(str(e))
-			return None
+			raise
 		except SessionError as e:
 			logging.debug(str(e))
-			return None
+			raise
 		except AssertionError as e:
 			logging.debug(str(e))
-			return None
+			raise
 
 	def init_samr_session(self):
 		if not self.samr:

diff --git a/powerview/utils/constants.py b/powerview/utils/constants.py
@@ -554,6 +554,8 @@ def to_list(self):
 # Universal SIDs
 WELL_KNOWN_SIDS = {
 	'S-1-0': 'Null Authority',
+	'S-1-15-2-1': 'ALL_APP_PACKAGES',
+	'S-1-15-2-2': 'ALL_RESTRICTED_APP_PACKAGES',
 	'S-1-0-0': 'Nobody',
 	'S-1-1': 'World Authority',
 	'S-1-1-0': 'Everyone',
@@ -627,6 +629,7 @@ def to_list(self):
 	'S-1-5-32-578': 'BUILTIN\\Hyper-V Administrators',
 	'S-1-5-32-579': 'BUILTIN\\Access Control Assistance Operators',
 	'S-1-5-32-580': 'BUILTIN\\Remote Management Users',
+	'S-1-5-80-956008885-3418522649-1831038044-1853292631-2271478464': 'TrustedInstaller',
 }
 
 def resolve_WellKnownSID(identifier):

diff --git a/powerview/utils/shell.py b/powerview/utils/shell.py
@@ -47,9 +47,9 @@ def get_prompt(init_proto, server_dns, cur_user, nameserver, target_domain=None,
 			  f'{bcolors.WARNING}{bcolors.BOLD}{init_proto}{bcolors.ENDC}'
 			  f'{bcolors.OKBLUE}─[{bcolors.ENDC}{bcolors.OKCYAN}{server_dns}{bcolors.ENDC}{bcolors.OKBLUE}]{bcolors.ENDC}'
 			  f'{bcolors.OKBLUE}─[{bcolors.ENDC}{cur_user}{bcolors.OKBLUE}]{bcolors.ENDC}'
+			  f'{bcolors.OKBLUE}-[NS:{nameserver if nameserver else "<auto>"}]{bcolors.ENDC}'
 			  f'{mcp_indicator}'
 			  f'{web_indicator}'
-			  f'{bcolors.OKBLUE} [NS:{nameserver if nameserver else "<auto>"}]{bcolors.ENDC}'
 			  f'{domain_indicator}'
 			  f'{cache_indicator}'
 			  f'\n{bcolors.OKBLUE}╰─{bcolors.BOLD}PV{bcolors.ENDC} {bcolors.OKGREEN}❯{bcolors.ENDC} ')