mirror of
https://github.com/cisagov/snafflepy
synced 2026-09-24 18:22:23 +00:00
added some files with "skeleton" code
This commit is contained in:
@@ -0,0 +1,77 @@
|
||||
from queue import Queue
|
||||
from enum import Flag
|
||||
|
||||
|
||||
class AdData:
|
||||
def __init__(self):
|
||||
self.domain_computers = []
|
||||
self.domain_users = []
|
||||
self.dfs_shares_dict = {}
|
||||
self.dfs_namespace_paths = []
|
||||
self.current_domain = None
|
||||
self.domain_name = ""
|
||||
self.target_domain = ""
|
||||
self.target_dc = ""
|
||||
self.target_domain_netbios_name = ""
|
||||
self.directory_search = None
|
||||
self.mq = Queue()
|
||||
|
||||
self.DirectoryContext = None
|
||||
self.DomainControllers = []
|
||||
|
||||
def get_domain_computers(self):
|
||||
return self.domain_computers
|
||||
|
||||
def get_domain_users(self):
|
||||
return self.domain_users
|
||||
|
||||
def get_dfs_shares_dict(self):
|
||||
return self.dfs_shares_dict
|
||||
|
||||
def get_dfs_namespace_paths(self):
|
||||
return self.dfs_namespace_paths
|
||||
|
||||
def get_directory_search(self):
|
||||
if self.directory_search is None:
|
||||
# set_directory_search()
|
||||
pass
|
||||
return self.directory_search
|
||||
|
||||
def get_netbios_domain_name(self):
|
||||
return
|
||||
|
||||
def set_directory_search(self):
|
||||
pass
|
||||
|
||||
def set_dfs_path(self):
|
||||
pass
|
||||
|
||||
def set_domain_computers(self):
|
||||
pass
|
||||
|
||||
def set_domain_users(self):
|
||||
pass
|
||||
|
||||
class UserAccountControlFlags(Flag):
|
||||
Script = 0x1
|
||||
AccountDisabled = 0x2
|
||||
HomeDirectoryRequired = 0x8
|
||||
AccountLockedOut = 0x10
|
||||
PasswordNotRequired = 0x20
|
||||
PasswordCannotChange = 0x40
|
||||
EncryptedTextPasswordAllowed = 0x80
|
||||
TempDuplicateAccount = 0x100
|
||||
NormalAccount = 0x200
|
||||
InterDomainTrustAccount = 0x800
|
||||
WorkstationTrustAccount = 0x1000
|
||||
ServerTrustAccount = 0x2000
|
||||
PasswordDoesNotExpire = 0x10000
|
||||
MnsLogonAccount = 0x20000
|
||||
SmartCardRequired = 0x40000
|
||||
TrustedForDelegation = 0x80000
|
||||
AccountNotDelegated = 0x100000
|
||||
UseDesKeyOnly = 0x200000
|
||||
DontRequirePreauth = 0x400000
|
||||
PasswordExpired = 0x800000
|
||||
TrustedToAuthenticateForDelegation = 0x1000000
|
||||
NoAuthDataRequired = 0x2000000
|
||||
@@ -0,0 +1,48 @@
|
||||
# import queue
|
||||
# import ldap3
|
||||
|
||||
|
||||
class DirectorySearch:
|
||||
def __init__(self, domain_controller, domain_name, ldap_username, ldap_password, ldap_port, secure_ldap, is_falted):
|
||||
self.domain_controller = domain_controller
|
||||
self.domain_name = domain_name
|
||||
self.base_ldap_path = "DC=" + self.domain_name.replace(".", ",DC=")
|
||||
# TODO: figure out the active directory library equivalent for python
|
||||
self.domain = None
|
||||
self.ldap_username = ldap_username
|
||||
self.ldap_password: ldap_password
|
||||
self.ldap_port = ldap_port
|
||||
self.secure_ldap = secure_ldap
|
||||
self.is_falted = is_falted
|
||||
self.connection_pool = []
|
||||
self.domain_guid_map = {}
|
||||
|
||||
def get_domain_name(self):
|
||||
if self.domain_name is None:
|
||||
return
|
||||
return self.domain_name
|
||||
|
||||
async def get_one(self):
|
||||
return
|
||||
|
||||
# https://ldap3.readthedocs.io/en/latest/tutorial_searches.html
|
||||
def query_ldap(self):
|
||||
return
|
||||
|
||||
def ranged_retrieval_async(self):
|
||||
return
|
||||
|
||||
def get_attribute_from_guid(self, guid: str, name: str):
|
||||
return self.domain_guid_map.get(guid, name)
|
||||
|
||||
def set_domain_name(self):
|
||||
return
|
||||
|
||||
def get_global_catalog_connection(self):
|
||||
return
|
||||
|
||||
def get_ldap_connection(self):
|
||||
return
|
||||
|
||||
def create_search_request(self):
|
||||
return
|
||||
@@ -0,0 +1,50 @@
|
||||
# import queue
|
||||
import ldap3
|
||||
|
||||
|
||||
class DirectorySearch:
|
||||
def __init__(self, domain_controller, domain_name, ldap_username, ldap_password, ldap_port, secure_ldap, is_falted):
|
||||
self.domain_controller = domain_controller
|
||||
self.domain_name = domain_name
|
||||
self.base_ldap_path = "DC=" + self.domain_name.replace(".", ",DC=")
|
||||
# TODO: figure out the active directory library equivalent for python
|
||||
self.domain = None
|
||||
self.ldap_username = ldap_username
|
||||
self.ldap_password: ldap_password
|
||||
self.ldap_port = ldap_port
|
||||
self.secure_ldap = secure_ldap
|
||||
self.is_falted = is_falted
|
||||
self.connection_pool = []
|
||||
self.domain_guid_map = {}
|
||||
|
||||
def get_domain_name(self):
|
||||
if self.domain_name is None:
|
||||
return
|
||||
return self.domain_name
|
||||
|
||||
async def get_one(self):
|
||||
return
|
||||
|
||||
# https://ldap3.readthedocs.io/en/latest/tutorial_searches.html
|
||||
def query_ldap(self):
|
||||
return
|
||||
|
||||
def ranged_retrieval_async(self):
|
||||
return
|
||||
|
||||
def get_attribute_from_guid(self, guid: str, name: str):
|
||||
return self.domain_guid_map.get(guid, name)
|
||||
|
||||
def set_domain_name(self):
|
||||
return
|
||||
|
||||
def get_global_catalog_connection(self):
|
||||
return
|
||||
|
||||
def get_ldap_conenction(self):
|
||||
return
|
||||
|
||||
def create_search_request(self):
|
||||
return
|
||||
|
||||
|
||||
@@ -2,5 +2,6 @@ def main():
|
||||
print("I snaffled 'til the snafflin was done")
|
||||
|
||||
|
||||
# TODO: Create SnaffleRunner class with equivalent functions and call the Run() method here.
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
||||
Reference in New Issue
Block a user