From d95e28410baf09e10c16f858c54656f7cdad180a Mon Sep 17 00:00:00 2001 From: ozgen Date: Tue, 3 Nov 2026 08:43:34 +0100 Subject: [PATCH] add: support for multiple credentials in target creation and modification --- gvm/protocols/gmp/_gmpnext.py | 100 +++++++---- gvm/protocols/gmp/requests/next/_targets.py | 73 ++++++++ .../entities/targets/test_create_target.py | 166 ++++++++++++++++++ .../entities/targets/test_modify_target.py | 137 +++++++++++++++ 4 files changed, 444 insertions(+), 32 deletions(-) diff --git a/gvm/protocols/gmp/_gmpnext.py b/gvm/protocols/gmp/_gmpnext.py index 7e21c966..e77c9d84 100644 --- a/gvm/protocols/gmp/_gmpnext.py +++ b/gvm/protocols/gmp/_gmpnext.py @@ -1667,28 +1667,40 @@ def create_target( reverse_lookup_unify: bool | None = None, port_range: str | None = None, port_list_id: EntityID | None = None, + ssh_credentials: list[tuple[EntityID, int | str]] | None = None, + ssh_elevate_credentials: list[EntityID] | None = None, + smb_credentials: list[EntityID] | None = None, + esxi_credentials: list[EntityID] | None = None, + snmp_credentials: list[EntityID] | None = None, + krb5_credentials: list[EntityID] | None = None, ) -> T: - """Create a new target + """Create a new target. Args: - name: Name of the target - asset_hosts_filter: Filter to select target host from assets hosts - hosts: List of hosts addresses to scan - exclude_hosts: List of hosts addresses to exclude from scan - comment: Comment for the target - ssh_credential_id: UUID of a ssh credential to use on target - ssh_credential_port: The port to use for ssh credential - smb_credential_id: UUID of a smb credential to use on target - snmp_credential_id: UUID of a snmp credential to use on target - esxi_credential_id: UUID of a esxi credential to use on target - alive_test: Which alive test to use + name: Name of the target. + asset_hosts_filter: Filter to select target hosts from assets. + hosts: List of host addresses to scan. + exclude_hosts: List of host addresses to exclude. + comment: Comment for the target. + ssh_credential_id: UUID of an SSH credential. + ssh_credential_port: Port for the SSH credential. + smb_credential_id: UUID of an SMB credential. + esxi_credential_id: UUID of an ESXi credential. + snmp_credential_id: UUID of an SNMP credential. + alive_test: Which alive test to use. allow_simultaneous_ips: Whether to scan multiple IPs of the - same host simultaneously - reverse_lookup_only: Whether to scan only hosts that have names - reverse_lookup_unify: Whether to scan only one IP when multiple IPs - have the same name. - port_range: Port range for the target - port_list_id: UUID of the port list to use on target + same host simultaneously. + reverse_lookup_only: Whether to scan only hosts with names. + reverse_lookup_unify: Whether to scan only one IP when + multiple IPs have the same name. + port_range: Port range for the target. + port_list_id: UUID of the port list. + ssh_credentials: List of SSH credential UUID and port pairs. + ssh_elevate_credentials: List of SSH elevate credential UUIDs. + smb_credentials: List of SMB credential UUIDs. + esxi_credentials: List of ESXi credential UUIDs. + snmp_credentials: List of SNMP credential UUIDs. + krb5_credentials: List of Kerberos credential UUIDs. """ return self._send_request_and_transform_response( Targets.create_target( @@ -1708,6 +1720,12 @@ def create_target( reverse_lookup_unify=reverse_lookup_unify, port_range=port_range, port_list_id=port_list_id, + ssh_credentials=ssh_credentials, + ssh_elevate_credentials=ssh_elevate_credentials, + smb_credentials=smb_credentials, + esxi_credentials=esxi_credentials, + snmp_credentials=snmp_credentials, + krb5_credentials=krb5_credentials, ) ) @@ -1729,27 +1747,39 @@ def modify_target( reverse_lookup_only: bool | None = None, reverse_lookup_unify: bool | None = None, port_list_id: EntityID | None = None, + ssh_credentials: list[tuple[EntityID, int | str]] | None = None, + ssh_elevate_credentials: list[EntityID] | None = None, + smb_credentials: list[EntityID] | None = None, + esxi_credentials: list[EntityID] | None = None, + snmp_credentials: list[EntityID] | None = None, + krb5_credentials: list[EntityID] | None = None, ) -> T: """Modify an existing target. Args: - target_id: UUID of target to modify. - comment: Comment on target. - name: Name of target. + target_id: UUID of the target to modify. + name: Name of the target. + comment: Comment on the target. hosts: List of target hosts. - exclude_hosts: A list of hosts to exclude. - ssh_credential_id: UUID of SSH credential to use on target. - ssh_credential_port: The port to use for ssh credential - smb_credential_id: UUID of SMB credential to use on target. - esxi_credential_id: UUID of ESXi credential to use on target. - snmp_credential_id: UUID of SNMP credential to use on target. - port_list_id: UUID of port list describing ports to scan. + exclude_hosts: List of hosts to exclude. + ssh_credential_id: UUID of an SSH credential. + ssh_credential_port: Port for the SSH credential. + smb_credential_id: UUID of an SMB credential. + esxi_credential_id: UUID of an ESXi credential. + snmp_credential_id: UUID of an SNMP credential. alive_test: Which alive tests to use. allow_simultaneous_ips: Whether to scan multiple IPs of the - same host simultaneously - reverse_lookup_only: Whether to scan only hosts that have names. - reverse_lookup_unify: Whether to scan only one IP when multiple IPs - have the same name. + same host simultaneously. + reverse_lookup_only: Whether to scan only hosts with names. + reverse_lookup_unify: Whether to scan only one IP when + multiple IPs have the same name. + port_list_id: UUID of the port list. + ssh_credentials: List of SSH credential UUID and port pairs. + ssh_elevate_credentials: List of SSH elevate credential UUIDs. + smb_credentials: List of SMB credential UUIDs. + esxi_credentials: List of ESXi credential UUIDs. + snmp_credentials: List of SNMP credential UUIDs. + krb5_credentials: List of Kerberos credential UUIDs. """ return self._send_request_and_transform_response( Targets.modify_target( @@ -1768,6 +1798,12 @@ def modify_target( reverse_lookup_only=reverse_lookup_only, reverse_lookup_unify=reverse_lookup_unify, port_list_id=port_list_id, + ssh_credentials=ssh_credentials, + ssh_elevate_credentials=ssh_elevate_credentials, + smb_credentials=smb_credentials, + esxi_credentials=esxi_credentials, + snmp_credentials=snmp_credentials, + krb5_credentials=krb5_credentials, ) ) diff --git a/gvm/protocols/gmp/requests/next/_targets.py b/gvm/protocols/gmp/requests/next/_targets.py index 16f843f2..6a07276b 100644 --- a/gvm/protocols/gmp/requests/next/_targets.py +++ b/gvm/protocols/gmp/requests/next/_targets.py @@ -59,6 +59,35 @@ def from_string( class Targets: + @staticmethod + def _add_multiple_credentials( + cmd: XmlCommand, + *, + ssh_credentials: list[tuple[EntityID, int | str]] | None = None, + ssh_elevate_credentials: list[EntityID] | None = None, + smb_credentials: list[EntityID] | None = None, + esxi_credentials: list[EntityID] | None = None, + snmp_credentials: list[EntityID] | None = None, + krb5_credentials: list[EntityID] | None = None, + ) -> None: + """Append a GMP element for each supplied target credential.""" + for credential_id, port in ssh_credentials or []: + xml_ssh = cmd.add_element( + "ssh_credential", attrs={"id": str(credential_id)} + ) + xml_ssh.add_element("port", str(port)) + + credential_lists = ( + ("ssh_elevate_credential", ssh_elevate_credentials), + ("smb_credential", smb_credentials), + ("esxi_credential", esxi_credentials), + ("snmp_credential", snmp_credentials), + ("krb5_credential", krb5_credentials), + ) + for element_name, credential_ids in credential_lists: + for credential_id in credential_ids or []: + cmd.add_element(element_name, attrs={"id": str(credential_id)}) + @classmethod def create_target( cls, @@ -73,6 +102,12 @@ def create_target( smb_credential_id: EntityID | None = None, esxi_credential_id: EntityID | None = None, snmp_credential_id: EntityID | None = None, + ssh_credentials: list[tuple[EntityID, int | str]] | None = None, + ssh_elevate_credentials: list[EntityID] | None = None, + smb_credentials: list[EntityID] | None = None, + esxi_credentials: list[EntityID] | None = None, + snmp_credentials: list[EntityID] | None = None, + krb5_credentials: list[EntityID] | None = None, alive_test: str | AliveTest | AliveTestV224 | None = None, allow_simultaneous_ips: bool | None = None, reverse_lookup_only: bool | None = None, @@ -92,6 +127,12 @@ def create_target( ssh_credential_port: The port to use for ssh credential smb_credential_id: UUID of a smb credential to use on target snmp_credential_id: UUID of a snmp credential to use on target + ssh_credentials: List of (SSH credential UUID, port) pairs. + ssh_elevate_credentials: List of SSH elevate credential UUIDs. + smb_credentials: List of SMB credential UUIDs. + esxi_credentials: List of ESXi credential UUIDs. + snmp_credentials: List of SNMP credential UUIDs. + krb5_credentials: List of Kerberos credential UUIDs. esxi_credential_id: UUID of a esxi credential to use on target alive_test: Which alive test to use allow_simultaneous_ips: Whether to scan multiple IPs of the @@ -150,6 +191,16 @@ def create_target( "snmp_credential", attrs={"id": str(snmp_credential_id)} ) + cls._add_multiple_credentials( + cmd, + ssh_credentials=ssh_credentials, + ssh_elevate_credentials=ssh_elevate_credentials, + smb_credentials=smb_credentials, + esxi_credentials=esxi_credentials, + snmp_credentials=snmp_credentials, + krb5_credentials=krb5_credentials, + ) + if alive_test: if not isinstance(alive_test, AliveTest): alive_test = AliveTest(alive_test) @@ -190,6 +241,12 @@ def modify_target( smb_credential_id: EntityID | None = None, esxi_credential_id: EntityID | None = None, snmp_credential_id: EntityID | None = None, + ssh_credentials: list[tuple[EntityID, int | str]] | None = None, + ssh_elevate_credentials: list[EntityID] | None = None, + smb_credentials: list[EntityID] | None = None, + esxi_credentials: list[EntityID] | None = None, + snmp_credentials: list[EntityID] | None = None, + krb5_credentials: list[EntityID] | None = None, alive_test: str | AliveTest | AliveTestV224 | None = None, allow_simultaneous_ips: bool | None = None, reverse_lookup_only: bool | None = None, @@ -209,6 +266,12 @@ def modify_target( smb_credential_id: UUID of SMB credential to use on target. esxi_credential_id: UUID of ESXi credential to use on target. snmp_credential_id: UUID of SNMP credential to use on target. + ssh_credentials: List of (SSH credential UUID, port) pairs. + ssh_elevate_credentials: List of SSH elevate credential UUIDs. + smb_credentials: List of SMB credential UUIDs. + esxi_credentials: List of ESXi credential UUIDs. + snmp_credentials: List of SNMP credential UUIDs. + krb5_credentials: List of Kerberos credential UUIDs. port_list_id: UUID of port list describing ports to scan. alive_test: Which alive tests to use. allow_simultaneous_ips: Whether to scan multiple IPs of the @@ -267,6 +330,16 @@ def modify_target( "snmp_credential", attrs={"id": str(snmp_credential_id)} ) + cls._add_multiple_credentials( + cmd, + ssh_credentials=ssh_credentials, + ssh_elevate_credentials=ssh_elevate_credentials, + smb_credentials=smb_credentials, + esxi_credentials=esxi_credentials, + snmp_credentials=snmp_credentials, + krb5_credentials=krb5_credentials, + ) + if allow_simultaneous_ips is not None: cmd.add_element( "allow_simultaneous_ips", to_bool(allow_simultaneous_ips) diff --git a/tests/protocols/gmpnext/entities/targets/test_create_target.py b/tests/protocols/gmpnext/entities/targets/test_create_target.py index 3988a0be..ac37bebf 100644 --- a/tests/protocols/gmpnext/entities/targets/test_create_target.py +++ b/tests/protocols/gmpnext/entities/targets/test_create_target.py @@ -253,3 +253,169 @@ def test_create_target_with_host_discovery_ipv6_alive_test(self): b"Host Discovery IPv6" b"" ) + + def test_create_target_with_multiple_ssh_credentials(self): + self.gmp.create_target( + "foo", + hosts=["foo"], + ssh_credentials=[ + ("c1", 22), + ("c2", 23), + ], + ) + + self.connection.send.has_been_called_with( + b"" + b"foo" + b"foo" + b'' + b"22" + b"" + b'' + b"23" + b"" + b"" + ) + + def test_create_target_with_multiple_smb_credentials(self): + self.gmp.create_target( + "foo", hosts=["foo"], smb_credentials=["c1", "c2"] + ) + + self.connection.send.has_been_called_with( + b"" + b"foo" + b"foo" + b'' + b'' + b"" + ) + + def test_create_target_with_multiple_esxi_credentials(self): + self.gmp.create_target( + "foo", hosts=["foo"], esxi_credentials=["c1", "c2"] + ) + + self.connection.send.has_been_called_with( + b"" + b"foo" + b"foo" + b'' + b'' + b"" + ) + + def test_create_target_with_multiple_snmp_credentials(self): + self.gmp.create_target( + "foo", hosts=["foo"], snmp_credentials=["c1", "c2"] + ) + + self.connection.send.has_been_called_with( + b"" + b"foo" + b"foo" + b'' + b'' + b"" + ) + + def test_create_target_with_multiple_krb5_credentials(self): + self.gmp.create_target( + "foo", hosts=["foo"], krb5_credentials=["c1", "c2"] + ) + + self.connection.send.has_been_called_with( + b"" + b"foo" + b"foo" + b'' + b'' + b"" + ) + + def test_create_target_with_multiple_ssh_elevate_credentials(self): + self.gmp.create_target( + "foo", + hosts=["foo"], + ssh_elevate_credentials=["c1", "c2"], + ) + + self.connection.send.has_been_called_with( + b"" + b"foo" + b"foo" + b'' + b'' + b"" + ) + + def test_create_target_with_mixed_credentials(self): + self.gmp.create_target( + "foo", + hosts=["foo"], + ssh_credentials=[ + ("ssh1", 22), + ("ssh2", 23), + ], + smb_credentials=["smb1"], + esxi_credentials=["esxi1"], + snmp_credentials=["snmp1"], + krb5_credentials=["krb1"], + ssh_elevate_credentials=["elevate1"], + ) + + self.connection.send.has_been_called_with( + b"" + b"foo" + b"foo" + b'' + b"22" + b"" + b'' + b"23" + b"" + b'' + b'' + b'' + b'' + b'' + b"" + ) + + def test_create_target_with_single_and_multiple_credentials(self): + self.gmp.create_target( + "foo", + hosts=["foo"], + ssh_credential_id="old", + ssh_credential_port=22, + ssh_credentials=[("new", 23)], + ) + + self.connection.send.has_been_called_with( + b"" + b"foo" + b"foo" + b'' + b"22" + b"" + b'' + b"23" + b"" + b"" + ) + + def test_create_target_with_empty_credential_lists(self): + self.gmp.create_target( + "foo", + hosts=["foo"], + ssh_credentials=[], + smb_credentials=[], + esxi_credentials=[], + snmp_credentials=[], + krb5_credentials=[], + ssh_elevate_credentials=[], + ) + + self.connection.send.has_been_called_with( + b"foofoo" + ) diff --git a/tests/protocols/gmpnext/entities/targets/test_modify_target.py b/tests/protocols/gmpnext/entities/targets/test_modify_target.py index aae550fe..d8692ce4 100644 --- a/tests/protocols/gmpnext/entities/targets/test_modify_target.py +++ b/tests/protocols/gmpnext/entities/targets/test_modify_target.py @@ -231,3 +231,140 @@ def test_modify_target_with_host_discovery_ipv6_alive_test_as_string(self): b"Host Discovery IPv6" b"" ) + + def test_modify_target_with_multiple_ssh_credentials(self): + self.gmp.modify_target( + target_id="t1", + ssh_credentials=[ + ("c1", 22), + ("c2", 23), + ], + ) + + self.connection.send.has_been_called_with( + b'' + b'' + b"22" + b"" + b'' + b"23" + b"" + b"" + ) + + def test_modify_target_with_multiple_smb_credentials(self): + self.gmp.modify_target(target_id="t1", smb_credentials=["c1", "c2"]) + + self.connection.send.has_been_called_with( + b'' + b'' + b'' + b"" + ) + + def test_modify_target_with_multiple_esxi_credentials(self): + self.gmp.modify_target(target_id="t1", esxi_credentials=["c1", "c2"]) + + self.connection.send.has_been_called_with( + b'' + b'' + b'' + b"" + ) + + def test_modify_target_with_multiple_snmp_credentials(self): + self.gmp.modify_target(target_id="t1", snmp_credentials=["c1", "c2"]) + + self.connection.send.has_been_called_with( + b'' + b'' + b'' + b"" + ) + + def test_modify_target_with_multiple_krb5_credentials(self): + self.gmp.modify_target(target_id="t1", krb5_credentials=["c1", "c2"]) + + self.connection.send.has_been_called_with( + b'' + b'' + b'' + b"" + ) + + def test_modify_target_with_multiple_ssh_elevate_credentials(self): + self.gmp.modify_target( + target_id="t1", + ssh_elevate_credentials=["c1", "c2"], + ) + + self.connection.send.has_been_called_with( + b'' + b'' + b'' + b"" + ) + + def test_modify_target_with_mixed_credentials(self): + self.gmp.modify_target( + target_id="t1", + ssh_credentials=[ + ("ssh1", 22), + ("ssh2", 23), + ], + ssh_elevate_credentials=["elevate1"], + smb_credentials=["smb1"], + esxi_credentials=["esxi1"], + snmp_credentials=["snmp1"], + krb5_credentials=["krb1"], + ) + + self.connection.send.has_been_called_with( + b'' + b'' + b"22" + b"" + b'' + b"23" + b"" + b'' + b'' + b'' + b'' + b'' + b"" + ) + + def test_modify_target_with_single_and_multiple_credentials(self): + self.gmp.modify_target( + target_id="t1", + ssh_credential_id="old", + ssh_credential_port=22, + ssh_credentials=[("new", 23)], + ) + + self.connection.send.has_been_called_with( + b'' + b'' + b"22" + b"" + b'' + b"23" + b"" + b"" + ) + + def test_modify_target_with_empty_credential_lists(self): + self.gmp.modify_target( + target_id="t1", + ssh_credentials=[], + ssh_elevate_credentials=[], + smb_credentials=[], + esxi_credentials=[], + snmp_credentials=[], + krb5_credentials=[], + ) + + self.connection.send.has_been_called_with( + b'' + )