import json import os import socket import ssl import time from threading import Thread from unittest.mock import patch import demistomock as demisto import pytest from Active_Directory_Query import group_dn, main from IAMApiModule import * BASE_TEST_PARAMS = { "server_ip": "127.0.0.1", "secure_connection": "None", "page_size": "500", "credentials": {"identifier": "bad", "password": "bad"}, } RETURN_ERROR_TARGET = "Active_Directory_Query.return_error" def test_bad_host_no_ssl(mocker): """ Given: - Demisto.params() with an invalid server (host), and insecure connection type (None). When: - Running the 'main()' function. Then: - Verify that the expected error message was raised. """ params = BASE_TEST_PARAMS params["server_ip"] = "127.0.0." mocker.patch.object(demisto, "params", return_value=BASE_TEST_PARAMS) return_error_mock = mocker.patch(RETURN_ERROR_TARGET) # validate our mock of params assert demisto.params().get("server_ip") == "127.0.0." main() assert return_error_mock.call_count == 1 # call_args last call with a tuple of args list and kwargs err_msg = return_error_mock.call_args[0][0] assert len(err_msg) < 100 assert "Failed to access" in err_msg assert "invalid server address" in err_msg @pytest.mark.filterwarnings("ignore::ResourceWarning") def test_bad_ssl(mocker): """ Given: - Demisto.params() with an ssl connection type (SSL) and a server (host) that will cause an SSL socket error. When: - Running the 'main()' function. Then: - Verify that the expected error message was raised. """ params = BASE_TEST_PARAMS.copy() params["server_ip"] = "185.199.108.153" # disable-secrets-detection params["secure_connection"] = "SSL" params["port"] = 443 mocker.patch.object(demisto, "params", return_value=params) return_error_mock = mocker.patch(RETURN_ERROR_TARGET) mocker.patch.object(demisto, "info") # validate our mock of params assert demisto.params().get("secure_connection") == "SSL" main() assert return_error_mock.call_count == 1 # call_args last call with a tuple of args list and kwargs err_msg = return_error_mock.call_args[0][0] assert "Failed to access" in err_msg assert 'Try using: "Trust any certificate" option.' in err_msg def ssl_bad_socket_server(port): context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) # cert and keyfile generated with # openssl req -x509 -nodes -days 3000 -newkey rsa:2048 -keyout key.pem -out cert.pem try: context.load_cert_chain("test_data/cert.pem", "test_data/key.pem") with socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) as sock: sock.bind(("127.0.0.1", port)) sock.listen(5) with context.wrap_socket(sock, server_side=True) as ssock: try: conn, addr = ssock.accept() except ssl.SSLError as err: if "TLSV1_ALERT_UNKNOWN_CA" in str(err): # all is ok. client refused our cert return raise conn.recv(32) msg = b"THIS IS A TEST SERVER WHICH IGNORES PROTOCOL\n\n" for _x in range(10): msg += msg conn.send(msg) conn.shutdown(socket.SHUT_RDWR) conn.close() except Exception as ex: pytest.fail(f"Failed starting ssl_bad_socket_server: {ex}") raise @pytest.mark.filterwarnings("ignore::ResourceWarning") def test_faulty_server(mocker): port = 9638 t = Thread(target=ssl_bad_socket_server, args=(port,)) t.start() time.sleep(1) # wait for socket server to startup params = BASE_TEST_PARAMS.copy() params["server_ip"] = "127.0.0.1" # disable-secrets-detection params["secure_connection"] = "SSL" params["unsecure"] = True params["port"] = port mocker.patch.object(demisto, "params", return_value=params) return_error_mock = mocker.patch(RETURN_ERROR_TARGET) # validate our mock of params assert demisto.params().get("secure_connection") == "SSL" main() t.join(5) assert return_error_mock.call_count == 1 # call_args last call with a tuple of args list and kwargs err_msg = return_error_mock.call_args[0][0] assert len(err_msg) < 125 assert "Failed to access" in err_msg assert 'Try using: "Trust any certificate" option.' not in err_msg def test_ssl_custom_cert(mocker, request): ENV_KEY = "SSL_CERT_FILE" os.environ[ENV_KEY] = "test_data/cert.pem" def cleanup(): os.environ.pop(ENV_KEY) request.addfinalizer(cleanup) port = 9637 t = Thread(target=ssl_bad_socket_server, args=(port,)) t.start() time.sleep(1) # wait for socket server to startup params = BASE_TEST_PARAMS.copy() params["server_ip"] = "127.0.0.1" # disable-secrets-detection params["secure_connection"] = "SSL" params["port"] = port mocker.patch.object(demisto, "params", return_value=params) return_error_mock = mocker.patch(RETURN_ERROR_TARGET) # validate our mock of params assert demisto.params().get("secure_connection") == "SSL" main() t.join(5) assert return_error_mock.call_count == 1 # call_args last call with a tuple of args list and kwargs err_msg = return_error_mock.call_args[0][0] assert len(err_msg) < 100 assert "Failed to access" in err_msg def test_endpoint_entry(): """ Given: Custom attributes to filter the computer object entry. When: The function filters the computer object according to the custom attributes. Then: The function will return all the computer object entry because custom attributes contain '*'. """ from Active_Directory_Query import endpoint_entry custom_attributes_with_asterisk = endpoint_entry({"dn": "dn", "name": "name", "memberOf": "memberOf"}, ["*"]) assert custom_attributes_with_asterisk == {"Groups": "memberOf", "Hostname": "name", "ID": "dn", "Type": "AD"} def get_outputs_from_user_profile(user_profile): entry_context = user_profile.to_entry() outputs = entry_context.get("Contents") return outputs def mock_demisto_map_object(object, mapper_name, incident_type): email = object.get("email") email_prefix = email.split("@")[0] return { "cn": email_prefix, "mail": email, "sAMAccountName": email_prefix, "userPrincipalName": email_prefix, "ou": "OU=Americas,OU=Demisto", } def test_get_iam_user_profile(mocker): from Active_Directory_Query import get_iam_user_profile mocker.patch.object(demisto, "mapObject", side_effect=mock_demisto_map_object) user_profile = { "email": "test2@paloaltonetworks.com", "username": "test", "locationregion": "Americas", "olduserdata": {"email": "test@paloaltonetworks.com", "username": "test", "locationregion": "Americas"}, } _, ad_user, sam_account_name = get_iam_user_profile(user_profile, "mock_mapper_out") assert sam_account_name == "test" assert ad_user def test_update_user_iam__username_change(mocker): """ Given: A valid user profile with valid mapping When: Running the `create_user_iam` command Then: The user was created successfully in AD. """ import Active_Directory_Query add_args, add_kwargs = [], {} class ConnectionMocker: entries = [] result = {"controls": {"1.2.840.113556.1.4.319": {"value": {"cookie": ""}}}} def search(self, *args, **kwargs): return def add(self, *args, **kwargs): nonlocal add_args, add_kwargs add_args, add_kwargs = args, kwargs return True def modify(self, *args, **kwargs): return True def modify_dn(self, *args, **kwargs): return True Active_Directory_Query.connection = ConnectionMocker() args = { "user-profile": json.dumps( { "email": "test2@paloaltonetworks.com", "username": "test", "locationregion": "Americas", "olduserdata": {"email": "test@paloaltonetworks.com", "username": "test", "locationregion": "Americas"}, } ) } mocker.patch.object(demisto, "mapObject", side_effect=mock_demisto_map_object) mocker.patch("Active_Directory_Query.check_if_user_exists_by_attribute", return_value=True) mocker.patch("Active_Directory_Query.get_user_activity_by_samaccountname", return_value=True) mocker.patch("Active_Directory_Query.user_dn", return_value="mock_dn") user_profile = Active_Directory_Query.update_user_iam( default_base_dn="mock_base_dn", args=args, create_if_not_exists=False, mapper_out="mock_mapper_out", disabled_users_group_cn="mock_disabled_users_group_cn", ) outputs = get_outputs_from_user_profile(user_profile) assert outputs.get("action") == IAMActions.UPDATE_USER assert outputs.get("success") is True assert outputs.get("email") == "test2@paloaltonetworks.com" assert outputs.get("username") == "test2" def test_create_user_iam(mocker): """ Given: A valid user profile with valid mapping When: Running the `create_user_iam` command Then: The user was created successfully in AD. """ import Active_Directory_Query add_args, add_kwargs = [], {} class ConnectionMocker: entries = [] result = {"controls": {"1.2.840.113556.1.4.319": {"value": {"cookie": ""}}}} def search(self, *args, **kwargs): return def add(self, *args, **kwargs): nonlocal add_args, add_kwargs add_args, add_kwargs = args, kwargs return True Active_Directory_Query.connection = ConnectionMocker() args = {"user-profile": json.dumps({"email": "test@paloaltonetworks.com", "username": "test", "locationregion": "Americas"})} mocker.patch("Active_Directory_Query.check_if_user_exists_by_attribute", return_value=False) mocker.patch.object( IAMUserProfile, "map_object", return_value={ "cn": "test", "mail": "test@paloaltonetworks.com", "sAMAccountName": "test", "userPrincipalName": "test", "ou": "OU=Americas,OU=Demisto", }, ) user_profile = Active_Directory_Query.create_user_iam("", args, "mapper_out", "") outputs = get_outputs_from_user_profile(user_profile) assert outputs.get("action") == IAMActions.CREATE_USER assert outputs.get("success") is True assert outputs.get("active") is False assert outputs.get("email") == "test@paloaltonetworks.com" def test_unseccsseful_create_user_iam_missing_ou(mocker): """ Given: A valid user profile with missing ou in the mapping When: Running the `create_user_iam` command Then: - The user was not created in AD. - An error message was returned. """ import Active_Directory_Query add_args, add_kwargs = [], {} class ConnectionMocker: entries = [] result = {"controls": {"1.2.840.113556.1.4.319": {"value": {"cookie": ""}}}} def search(self, *args, **kwargs): return def add(self, *args, **kwargs): nonlocal add_args, add_kwargs add_args, add_kwargs = args, kwargs return True Active_Directory_Query.connection = ConnectionMocker() args = {"user-profile": json.dumps({"email": "test@paloaltonetworks.com", "username": "test", "locationregion": "Americas"})} mocker.patch("Active_Directory_Query.check_if_user_exists_by_attribute", return_value=False) mocker.patch.object( IAMUserProfile, "map_object", return_value={"cn": "test", "mail": "test@paloaltonetworks.com", "sAMAccountName": "test", "userPrincipalName": "test"}, ) user_profile = Active_Directory_Query.create_user_iam("", args, "mapper_out", "") outputs = get_outputs_from_user_profile(user_profile) assert outputs.get("action") == IAMActions.CREATE_USER assert outputs.get("success") is False assert outputs.get("email") == "test@paloaltonetworks.com" assert "User must have an Organizational Unit (OU)" in outputs.get("errorMessage") def test_unseccsseful_create_user_iam_missing_samaccountname(mocker): """ Given: A valid user profile with missing samaccountname in the mapping When: Running the `create_user_iam` command Then: - The user was not created in AD. - An error message was returned. """ import Active_Directory_Query add_args, add_kwargs = [], {} class ConnectionMocker: entries = [] result = {"controls": {"1.2.840.113556.1.4.319": {"value": {"cookie": ""}}}} def search(self, *args, **kwargs): return def add(self, *args, **kwargs): nonlocal add_args, add_kwargs add_args, add_kwargs = args, kwargs return True Active_Directory_Query.connection = ConnectionMocker() args = {"user-profile": json.dumps({"email": "test@paloaltonetworks.com", "username": "test", "locationregion": "Americas"})} mocker.patch("Active_Directory_Query.check_if_user_exists_by_attribute", return_value=False) mocker.patch.object( IAMUserProfile, "map_object", return_value={ "cn": "test", "mail": "test@paloaltonetworks.com", "ou": "OU=Americas,OU=Demisto", "userPrincipalName": "test", }, ) user_profile = Active_Directory_Query.create_user_iam("", args, "mapper_out", "") outputs = get_outputs_from_user_profile(user_profile) assert outputs.get("action") == IAMActions.CREATE_USER assert outputs.get("success") is False assert outputs.get("email") == "test@paloaltonetworks.com" assert "User must have a sAMAccountName" in outputs.get("errorMessage") def test_group_entry_no_custom_attributes(): """ Given: Custom attributes to filter the group object entry. When: The function filters the group object according to the custom attributes. Then: The function will return all the group object entry because custom attributes contain '*'. """ from Active_Directory_Query import group_entry custom_attributes_with_asterisk = group_entry({"dn": "dn", "name": "name", "memberOf": "memberOf"}, ["*"]) assert custom_attributes_with_asterisk == {"Groups": "memberOf", "ID": "dn", "Name": "name", "Type": "AD"} def test_group_entry(): """ Given: Custom attributes to filter the group object entry. When: The function filters the group object according to the custom attributes. Then: The function will return all the group object entry because custom attributes contain '*'. """ from Active_Directory_Query import group_entry custom_attributes_with_asterisk = group_entry( {"dn": "dn", "name": "name", "memberOf": "memberOf", "displayName": "display name"}, ["displayName"] ) assert custom_attributes_with_asterisk == { "Groups": "memberOf", "ID": "dn", "Name": "name", "Type": "AD", "displayName": "display name", } def test_search_group_members(mocker): """ sanity test for search_group_members method """ import Active_Directory_Query class EntryMocker: def entry_to_json(self): return '{"dn": "dn","attributes": {"memberOf": ["memberOf"], "name": ["name"]}}' class ConnectionMocker: entries = [EntryMocker()] result = {"controls": {"1.2.840.113556.1.4.319": {"value": {"cookie": b""}}}} def search(self, *args, **kwargs): time.sleep(1) expected_entry = { "ActiveDirectory.Groups(obj.dn ==dn)": {"dn": "dn", "members": [{"dn": "dn", "category": "group"}]}, "ActiveDirectory.Groups(obj.dn == val.dn)": [{"dn": "dn", "memberOf": ["memberOf"], "name": ["name"]}], "Group": [{"Type": "AD", "ID": "dn", "Name": ["name"], "Groups": ["memberOf"]}], "ActiveDirectory(true)": {"GroupsPageCookie": base64.b64encode(b"").decode("utf-8")}, } expected_results = { "ContentsFormat": "json", "Type": 1, "Contents": [{"dn": "dn", "attributes": {"memberOf": ["memberOf"], "name": ["name"]}}], "ReadableContentsFormat": "markdown", "HumanReadable": "### Active Directory - Get Group Members\n|" "dn|memberOf|name|\n|---|---|---|\n| dn | memberOf | name |\n", "EntryContext": expected_entry, } expected_results = f"demisto results: {json.dumps(expected_results, indent=4, sort_keys=True)}" mocker.patch.object(demisto, "args", return_value={"member-type": "group", "group-dn": "dn", "time_limit": "1"}) Active_Directory_Query.connection = ConnectionMocker() with patch("logging.Logger.info") as mock: Active_Directory_Query.search_group_members("dc", 1) mock.assert_called_with(expected_results) def test_group_dn_escape_characters(): """ Given: Group name with parentheses When: Running the function group_dn Then: The function search gets the group name after escape special characters. """ import Active_Directory_Query class EntryMocker: def entry_to_json(self): return '{"dn": "dn","attributes": {"memberOf": ["memberOf"], "name": ["name"]}}' class ConnectionMocker: entries = [EntryMocker()] result = {"controls": {"1.2.840.113556.1.4.319": {"value": {"cookie": ""}}}} Active_Directory_Query.connection = ConnectionMocker() with patch("Active_Directory_Query.search", return_value=[EntryMocker()]) as mock: group_dn("group(group)", "") mock.assert_called_with("(&(objectClass=group)(cn=group\\28group\\29))", "") def test_search__no_control_exist(mocker): """ Given: No control key in the result When: Run any search query Then: The result return 'no entries' instead of throw exception """ import Active_Directory_Query class ConnectionMocker: entries = [] result = {} def search(self, *args, **kwargs): return mocker.patch.object(demisto, "results") Active_Directory_Query.connection = ConnectionMocker() Active_Directory_Query.search_users("dc=test,dc=test_1", page_size=20) assert "**No entries.**" in demisto.results.call_args[0][0]["HumanReadable"] def test_search_attributes_to_exclude(mocker): """ Given: attributes_to_exclude When: Run any search query Then: The given arguments where excluded from human_readable and context_data """ import Active_Directory_Query class EntryMocker: def entry_to_json(self): return '{"dn": "dn"}' class ConnectionMocker: entries = [EntryMocker()] result = {"controls": {"1.2.840.113556.1.4.319": {"value": {"cookie": b""}}}} def search(self, *args, **kwargs): time.sleep(1) expected_results = { "ContentsFormat": "json", "Type": 1, "Contents": [{"dn": "dn"}], "ReadableContentsFormat": "markdown", "HumanReadable": "### Active Directory - Get Users\n|dn|\n|---|\n| dn |\n", "EntryContext": { "ActiveDirectory.Users(obj.dn == val.dn)": [{"dn": "dn"}], "Account(obj.ID == val.ID)": [ { "Type": "AD", "ID": "dn", "Email": None, "Username": None, "DisplayName": None, "Managr": None, "Manager": None, "Groups": None, } ], "ActiveDirectory(true)": {"UsersPageCookie": base64.b64encode(b"").decode("utf-8")}, }, } expected_results = f"demisto results: {json.dumps(expected_results, indent=4, sort_keys=True)}" mocker.patch.object( demisto, "args", return_value={ "attributes-to-exclude": "memberOf,name,mail,displayName,manager,sAMAccountName,userAccountControl", "page-size": "1", }, ) Active_Directory_Query.connection = ConnectionMocker() with patch("logging.Logger.info") as mock: Active_Directory_Query.search_users("dc", 1) mock.assert_called_with(expected_results) def test_user_account_to_boolean_fields(): """ Given: a userAccountControl value When: parsing the userAccountControl fields Then: Only the relevant fields will be marked as true """ import Active_Directory_Query fields = Active_Directory_Query.user_account_to_boolean_fields(0x50) assert {k for k, v in fields.items() if v} == {"LOCKOUT", "PASSWD_CANT_CHANGE"} @pytest.mark.parametrize("flags", [512, 0, 544]) def test_restore_user(mocker, flags): """ Given: A disabled user. When: Calling restore_user method. Then: Verify the existing flag is returned. """ from Active_Directory_Query import restore_user re_val = {"flat": [{"userAccountControl": [flags]}]} mocker.patch("Active_Directory_Query.search_with_paging", return_value=re_val) mocker.patch.object(demisto, "args") assert restore_user("test_user", 0) == flags def test_enable_user_with_restore_user_option(mocker): """ Given: A disabled user. When: Calling enable_user method. Then: Verify the existing flag is returned with the disable bit off. """ from Active_Directory_Query import enable_user disabled_account_with_properties = 546 enabled_account_with_properties = 544 mocker.patch("Active_Directory_Query.restore_user", return_value=disabled_account_with_properties) mocker.patch("Active_Directory_Query.user_dn", return_value="test_dn") modify_data = mocker.patch("Active_Directory_Query.modify_object") mocker.patch.object(demisto, "args") enable_user("test_user", 0) assert modify_data.call_args.args[1].get("userAccountControl")[0][1] == enabled_account_with_properties def test_search_with_paging_bug(mocker): """ Given: page size larger than 1. When: running get-group-members command. Then: time_limit results returned. """ import Active_Directory_Query class EntryMocker: def entry_to_json(self): return '{"dn": "dn","attributes": {"memberOf": ["memberOf"], "name": ["name"]}}' class ConnectionMocker: entries = [] result = {"controls": {"1.2.840.113556.1.4.319": {"value": {"cookie": b""}}}} def search(self, *args, **kwargs): page_size = kwargs.get("paged_size") if page_size: self.entries = [EntryMocker() for i in range(page_size)] time.sleep(1) mocker.patch.object(demisto, "results") mocker.patch.object(demisto, "args", return_value={"member-type": "group", "group-dn": "dn", "time_limit": "3"}) Active_Directory_Query.connection = ConnectionMocker() with patch("logging.Logger.info"): Active_Directory_Query.search_group_members("dc", 1) assert len(demisto.results.call_args[0][0]["Contents"]) == 3 def test_password_not_expire_missing_username(mocker): """ Given: A demisto args object with missing username and a valid value. When: running set_password_not_expire command. Then: Verify that a a missing username exception is raised. """ from Active_Directory_Query import set_password_not_expire mocker.patch.object(demisto, "args", return_value={"username": None, "value": True}) default_base_dn = {} with pytest.raises(Exception) as err: set_password_not_expire(default_base_dn) assert err.value.args[0] == "Missing argument - You must specify a username (sAMAccountName)." @pytest.mark.parametrize( "connection_type, unsecure, expected_auto_bind_value", [ ("Start TLS", True, "TLS_BEFORE_BIND"), ("Start TLS", False, "TLS_BEFORE_BIND"), ("TLS", False, "TLS_BEFORE_BIND"), ("TLS", True, "NO_TLS"), ("SSL", True, "NO_TLS"), ("SSL", False, "NO_TLS"), ("None", True, "NO_TLS"), ("None", False, "NO_TLS"), ], ) def test_get_auto_bind_value(connection_type, unsecure, expected_auto_bind_value): """ Given: - A connection type: 1. Start TLS 2. TLS 3. SSL 4. None When: - Running the 'get_auto_bind_value()' function. Then: - Verify that the returned auto_bind value is as expected: 1. 'TLS_BEFORE_BIND' - which means that connection should upgrade it's secure level to TLS before the bind itself (STARTTLS command is executed). 2. 'TLS_BEFORE_BIND' - for unsecure=False and 'NO_TLS' for unsecure=True 3. 'NO_TLS' - The connection is secured from the beginning, thus STARTTLS command shouldn't be executed. 4. 'NO_TLS' - Connection is insecure (cleartext) and shouldn't be upgraded to TLS. """ from Active_Directory_Query import get_auto_bind_value auto_bind_value = get_auto_bind_value(connection_type, unsecure) assert auto_bind_value == expected_auto_bind_value @pytest.mark.parametrize( "ssl_version, expected_ssl_version", [("TLS", 2), ("TLSv1", 3), ("TLSv1_1", 4), ("TLSv1_2", 5), ("TLS_CLIENT", 16), (None, None), ("None", None)], ) def test_get_ssl_version(ssl_version, expected_ssl_version): """ Given: - An ssl protocol version: 1. TLS 2. TLSv1 3. TLSv1_1 4. TLSv1_2 5. TLS_CLIENT 6. None 7. 'None' When: - Running the 'get_ssl_version()' function. Then: - Verify that the returned ssl version value is as expected: 1. TLS - 2 2. TLSv1 - 3 3. TLSv1_1 - 4 4. TLSv1_2 - 5 5. TLS_CLIENT - 16 6. None - None 7. 'None' - None """ from Active_Directory_Query import get_ssl_version ssl_version_value = get_ssl_version(ssl_version) assert ssl_version_value == expected_ssl_version def test_search_users_empty_userAccountControl(mocker): """ Given: The 'userAccountControl' attribute was returned empty When: Run the 'ad-get-user' command Then: The result returns without raise IndexError: list index out of range """ import Active_Directory_Query class EntryMocker: def entry_to_json(self): return ( '{"attributes": {"displayName": [], "mail": [], "manager": [], "memberOf": ["memberOf"], ' '"name": ["Guest"], "sAMAccountName": ["Guest"], "userAccountControl": []}, "dn": "test_dn"}' ) class ConnectionMocker: entries = [EntryMocker()] result = {"controls": {"1.2.840.113556.1.4.319": {"value": {"cookie": b""}}}} def search(self, *args, **kwargs): time.sleep(1) expected_results = { "ContentsFormat": "json", "Type": 1, "Contents": [ { "attributes": { "displayName": [], "mail": [], "manager": [], "memberOf": ["memberOf"], "name": ["Guest"], "sAMAccountName": ["Guest"], "userAccountControl": [], }, "dn": "test_dn", } ], "ReadableContentsFormat": "markdown", "HumanReadable": "### Active Directory - Get Users\n|displayName|dn|mail|manager|memberOf|name" "|sAMAccountName|userAccountControl|\n|---|---|---|---|---|---|---|---|\n| |" " test_dn | | | memberOf | Guest | Guest | |\n", "EntryContext": { "ActiveDirectory.Users(obj.dn == val.dn)": [ { "dn": "test_dn", "displayName": [], "mail": [], "manager": [], "memberOf": ["memberOf"], "name": ["Guest"], "sAMAccountName": ["Guest"], "userAccountControl": [], } ], "Account(obj.ID == val.ID)": [ { "Type": "AD", "ID": "test_dn", "Email": [], "Username": ["Guest"], "DisplayName": [], "Managr": [], "Manager": [], "Groups": ["memberOf"], } ], "ActiveDirectory(true)": {"UsersPageCookie": base64.b64encode(b"").decode("utf-8")}, }, } expected_results = f"demisto results: {json.dumps(expected_results, indent=4, sort_keys=True)}" mocker.patch.object(demisto, "args", return_value={"page-size": "1"}) Active_Directory_Query.connection = ConnectionMocker() with patch("logging.Logger.info") as mock: Active_Directory_Query.search_users("dc", 1) mock.assert_called_with(expected_results) def test_test_credentials_command(mocker): """ Given: A demisto args object with username and password When: Run the 'ad-test-credentials' command Then: The result returns with successful connection """ import Active_Directory_Query args = {"username": "username_test_credentials", "password": "password_test_credentials"} mocker.patch.object(demisto, "args", return_value=args) class MockConnection: def unbind(self): pass def mock_create_connection(server, server_ip, username, password, ntlm_connection, auto_bind): return MockConnection() with ( patch("Active_Directory_Query.create_connection", side_effect=mock_create_connection), patch("Active_Directory_Query.Connection.unbind", side_effect=MockConnection.unbind), ): command_results = Active_Directory_Query.test_credentials_command( BASE_TEST_PARAMS["server_ip"], "server", ntlm_connection="true", auto_bind="auto_bind" ) assert command_results.readable_output == "Credential test with username username_test_credentials succeeded." @pytest.mark.parametrize( "dn,expected", [ ("CN=name, lastname,OU=Test1,DC=dc1,DC=dc2", "CN=name, lastname"), ("CN=name\\ lastname,OU=Test1,DC=dc1,DC=dc2", "CN=name lastname"), ("CN=name,DC=dc1,DC=dc2", "CN=name"), ], ) def test_modify_user_ou(mocker, dn, expected): """ Given: - user with CN contains // - user with CN contains comma - user without ou When: Run the 'ad-modify-ou' command Then: Validate the cn extracted as expected """ import Active_Directory_Query class MockConnection: def modify_dn(self, dn, cn, new_superior): pass Active_Directory_Query.connection = MockConnection() new_ou = "OU=Test2" connection_mocker = mocker.patch.object(Active_Directory_Query.connection, "modify_dn", return_value=True) Active_Directory_Query.modify_user_ou(dn, new_ou) assert connection_mocker.call_args[0][1] == expected def test_search_users_with_msDSUserAccountControlComputed(mocker): """ Given: The 'msDSUserAccountControlComputed' was returned. When: Run the 'ad-get-user' command Then: The user_account_to_boolean_fields_msDS_user_account_control_computed was called. """ import Active_Directory_Query class EntryMocker: def entry_to_json(self): return ( '{"attributes": {"displayName": [], "mail": [], "manager": [], "memberOf": ["memberOf"], ' '"name": ["Guest"], "sAMAccountName": ["Guest"], "userAccountControl": [0], \ "msDS-User-Account-Control-Computed": [0]},"dn": "test_dn"}' ) class ConnectionMocker: entries = [EntryMocker()] result = {"controls": {"": {"value": {"cookie": b""}}}} def search(self, *args, **kwargs): time.sleep(1) mocker.patch.object(demisto, "args", return_value={"page-size": "1"}) mocker.patch.object(demisto, "results") mocker_msDSUserAccountControlComputed = mocker.patch.object( Active_Directory_Query, "user_account_to_boolean_fields_msDS_user_account_control_computed", return_value={"PASSWORD_EXPIRED": True, "LOCKOUT": True}, ) Active_Directory_Query.connection = ConnectionMocker() Active_Directory_Query.search_users("dc", 1) mocker_msDSUserAccountControlComputed.assert_called_once() assert "msDS-User-Account-Control-Computed" in demisto.results.call_args[0][0]["Contents"][0].get("attributes") assert ( demisto.results.call_args[0][0]["EntryContext"] .get("ActiveDirectory.Users(obj.dn == val.dn)", {})[0] .get("userAccountControlFields") .get("PASSWORD_EXPIRED") is True ) def test_base_dn_verified_success(mocker): """ Given: A valid base DN and a successful LDAP connection. When: Running the 'base_dn_verified' function. Then: Verify that the function returns True and uses BASE scope with optimized parameters. """ import Active_Directory_Query from ldap3 import BASE search_args = [] search_kwargs = {} class ConnectionMocker: result = {"description": "success"} def search(self, *args, **kwargs): nonlocal search_args, search_kwargs search_args = args search_kwargs = kwargs return True Active_Directory_Query.connection = ConnectionMocker() mocker.patch.object(demisto, "info") result = Active_Directory_Query.base_dn_verified("dc=example,dc=com") # Verify the function returns True assert result is True # Verify the search was called with optimized parameters assert search_kwargs.get("search_base") == "dc=example,dc=com" assert search_kwargs.get("search_filter") == "(objectClass=*)" assert search_kwargs.get("search_scope") == BASE # Verify BASE scope is used assert search_kwargs.get("size_limit") == 1 # Verify size limit is 1 assert search_kwargs.get("attributes") == ["1.1"] # Verify no attributes are fetched def test_base_dn_verified_failure(mocker): """ Given: An invalid base DN that causes the LDAP search to fail. When: Running the 'base_dn_verified' function. Then: Verify that the function returns False and logs the failure. """ import Active_Directory_Query class ConnectionMocker: result = {"description": "noSuchObject"} def search(self, *args, **kwargs): return False Active_Directory_Query.connection = ConnectionMocker() info_mock = mocker.patch.object(demisto, "info") result = Active_Directory_Query.base_dn_verified("dc=invalid,dc=com") # Verify the function returns False assert result is False # Verify that the failure was logged assert info_mock.call_count == 1 assert "Base DN verification failed" in info_mock.call_args[0][0] def test_base_dn_verified_exception(mocker): """ Given: A base DN that causes an exception during LDAP search. When: Running the 'base_dn_verified' function. Then: Verify that the function returns False and logs the error. """ import Active_Directory_Query class ConnectionMocker: def search(self, *args, **kwargs): raise Exception("Connection timeout") Active_Directory_Query.connection = ConnectionMocker() error_mock = mocker.patch.object(demisto, "error") result = Active_Directory_Query.base_dn_verified("dc=example,dc=com") # Verify the function returns False assert result is False # Verify that the error was logged assert error_mock.call_count == 1 assert "Error during Base DN verification" in error_mock.call_args[0][0] assert "Connection timeout" in error_mock.call_args[0][0] # ===== Tests for prepare_attribute_value ===== @pytest.mark.parametrize( "attribute_value, attribute_type, expected", [ pytest.param("displayName", None, "displayName", id="no-type-returns-string"), pytest.param("some value", "", "some value", id="empty-type-returns-string"), pytest.param( "0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0", "byte", bytes(21), id="byte-type-21-zeros-logonHours-full-lockout", ), pytest.param("0, 255, 128", "byte", bytes([0, 255, 128]), id="byte-type-with-whitespace"), pytest.param("1,2,3", "BYTE", bytes([1, 2, 3]), id="byte-type-case-insensitive"), ], ) def test_prepare_attribute_value_success(attribute_value, attribute_type, expected): """ Given: A valid attribute value and an optional attribute type. When: Calling prepare_attribute_value. Then: The correct Python object is returned (bytes for 'byte' type, str otherwise). """ from Active_Directory_Query import prepare_attribute_value result = prepare_attribute_value(attribute_value, attribute_type) assert result == expected @pytest.mark.parametrize( "attribute_value, attribute_type, expected_error_fragment", [ pytest.param( "[byte[]](0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0)", "byte", "cannot be parsed as a comma-separated list of integers", id="byte-type-powershell-syntax-rejected", ), pytest.param("0,256,0", "byte", "out-of-range integers", id="byte-type-value-above-255"), pytest.param("0,-1,0", "byte", "out-of-range integers", id="byte-type-negative-value"), pytest.param(None, "byte", "attribute-value must be provided", id="byte-type-none-value"), pytest.param("", "byte", "attribute-value must be provided", id="byte-type-empty-string"), ], ) def test_prepare_attribute_value_errors(attribute_value, attribute_type, expected_error_fragment): """ Given: An invalid attribute value for the given attribute type. When: Calling prepare_attribute_value. Then: A DemistoException is raised with a descriptive message. """ from CommonServerPython import DemistoException from Active_Directory_Query import prepare_attribute_value with pytest.raises(DemistoException, match=expected_error_fragment): prepare_attribute_value(attribute_value, attribute_type) def test_update_user_with_byte_attribute(mocker): """ Given: The ad-update-user command is called with attribute-type='byte' and a comma-separated list of integers representing logonHours (21 zero bytes = full lockout). When: update_user() is executed. Then: The LDAP modify call receives a raw bytes object (not a string) as the attribute value, which is what the AD server requires for Octet String attributes. """ import Active_Directory_Query class MockConnection: result = {"description": "success", "result": 0} last_dn: str = "" last_modification: dict = {} def modify(self, dn, modification): self.last_dn = dn self.last_modification = modification self.result = {"description": "success", "result": 0} return True mock_conn = MockConnection() Active_Directory_Query.connection = mock_conn # Mock user_dn so we skip the LDAP search entirely mocker.patch.object( Active_Directory_Query, "user_dn", return_value="CN=Test User,DC=example,DC=com", ) mocker.patch.object( demisto, "args", return_value={ "username": "testuser", "attribute-name": "logonHours", "attribute-value": "0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0", "attribute-type": "byte", }, ) mocker.patch.object(demisto, "results") Active_Directory_Query.update_user("dc=example,dc=com") # The value sent to LDAP must be raw bytes, not a string sent_value = mock_conn.last_modification["logonHours"][0][1] assert isinstance(sent_value, bytes), f"Expected bytes, got {type(sent_value)}" assert sent_value == bytes(21)