feat: allow recreating ssh key when public key in the API does not match (#634)

##### SUMMARY

- Log a warning when the provided public key does not match one in the
API.
- When the public key does not match the one in the API, allow
recreating the SSH Key in the API using the ``force=true`` argument.


Closes: #578 

##### ISSUE TYPE
- Feature Pull Request

##### COMPONENT NAME
<!--- Write the short name of the module, plugin, task or feature below
-->
`ssh_key`

##### ADDITIONAL INFORMATION
In Hetzner API, we do not have any public_key change endpoint and only
updating names and labels are allowed.
For public_key The only way is removing and re-creating. `force` option
allows users to do re-creation if needed.

---------

Co-authored-by: jo <ljonas@riseup.net>

changelogs/fragments/ssh-key-force-argument.yml

@@ -0,0 +1,4 @@
+minor_changes:
+  - ssh_key - Log a warning when the provided public key does not match one in the API.
+  - ssh_key - When the public key does not match the one in the API, allow recreating the
+    SSH Key in the API using the ``force=true`` argument.

plugins/module_utils/ssh.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from base64 import b64decode
+from hashlib import md5
+
+
+def ssh_public_key_md5_fingerprint(value: str) -> str:
+    """
+    Compute the md5 fingerprint of a SSH public key.
+    """
+    parts = value.strip().split()
+    if len(parts) < 2:
+        raise ValueError("invalid ssh public key")
+
+    raw = b64decode(parts[1].encode("ascii"))
+    digest = md5(raw).hexdigest()
+
+    return ":".join(a + b for a, b in zip(digest[::2], digest[1::2]))

plugins/modules/ssh_key.py

@@ -44,6 +44,11 @@ options:
             - The Public Key to add.
             - Required if ssh_key does not exist.
         type: str
+    force:
+        description:
+            - Recreate the SSH Key if the public key does not match the one in the API.
+        type: bool
+        default: false
     state:
         description:
             - State of the ssh_key.
@@ -115,6 +120,7 @@ hcloud_ssh_key:
 from ansible.module_utils.basic import AnsibleModule
 
 from ..module_utils.hcloud import AnsibleHCloud
+from ..module_utils.ssh import ssh_public_key_md5_fingerprint
 from ..module_utils.vendor.hcloud import HCloudException
 from ..module_utils.vendor.hcloud.ssh_keys import BoundSSHKey
 
@@ -175,6 +181,21 @@ class AnsibleHCloudSSHKey(AnsibleHCloud):
                 self.hcloud_ssh_key.update(labels=labels)
             self._mark_as_changed()
 
+        public_key = self.module.params.get("public_key")
+        if public_key is not None:
+            fingerprint = ssh_public_key_md5_fingerprint(public_key)
+            if fingerprint != self.hcloud_ssh_key.fingerprint:
+                if self.module.params.get("force"):
+                    if not self.module.check_mode:
+                        self.hcloud_ssh_key.delete()
+                        self._create_ssh_key()
+                    self._mark_as_changed()
+                else:
+                    self.module.warn(
+                        f"SSH Key '{self.hcloud_ssh_key.name}' in the API has a "
+                        f"different public key than the one provided. "
+                        f"Use the force=true argument to recreate the SSH Key in the API."
+                    )
         self._get_ssh_key()
 
     def present_ssh_key(self):
@@ -204,6 +225,7 @@ class AnsibleHCloudSSHKey(AnsibleHCloud):
                 public_key={"type": "str"},
                 fingerprint={"type": "str"},
                 labels={"type": "dict"},
+                force={"type": "bool", "default": False},
                 state={
                     "choices": ["absent", "present"],
                     "default": "present",

tests/integration/targets/setup_ssh_keypair/tasks/main.yml

@@ -17,3 +17,15 @@
     path: "{{ _tmp_ssh_key_file.path }}"
     force: true
   register: test_ssh_keypair
+
+- name: Create temporary file for test_ssh_keypair2
+  ansible.builtin.tempfile:
+    path: ~/tmp
+    suffix: "{{ hcloud_ssh_key_name }}"
+  register: _tmp_ssh_key_file2
+
+- name: Create test_ssh_keypair2
+  community.crypto.openssh_keypair:
+    path: "{{ _tmp_ssh_key_file2.path }}"
+    force: true
+  register: test_ssh_keypair2

tests/integration/targets/ssh_key/tasks/test.yml

@@ -93,7 +93,7 @@
       key: value
       test: "val123"
   register: result
-- name: test update ssh key  with other labels
+- name: test update ssh key with other labels
   assert:
     that:
       - result is changed
@@ -136,9 +136,45 @@
       - result is failed
       - result.failure.code == "uniqueness_error"
 
+- name: test update public key warning
+  hetzner.hcloud.ssh_key:
+    name: "{{ hcloud_ssh_key_name }}"
+    public_key: "{{ test_ssh_keypair2.public_key }}"
+  register: result
+- name: verify update public key warning
+  assert:
+    that:
+      - result is not changed
+      - result.hcloud_ssh_key.name == hcloud_ssh_key_name
+      - result.hcloud_ssh_key.public_key == test_ssh_keypair.public_key
+
+- name: test update public key with force
+  hetzner.hcloud.ssh_key:
+    name: "{{ hcloud_ssh_key_name }}"
+    public_key: "{{ test_ssh_keypair2.public_key }}"
+    force: true
+  register: result
+- name: verify update public key with force
+  assert:
+    that:
+      - result is changed
+      - result.hcloud_ssh_key.name == hcloud_ssh_key_name
+      - result.hcloud_ssh_key.public_key == test_ssh_keypair2.public_key
+
+- name: test update public key with force idempotence
+  hetzner.hcloud.ssh_key:
+    name: "{{ hcloud_ssh_key_name }}"
+    public_key: "{{ test_ssh_keypair2.public_key }}"
+    force: true
+  register: result
+- name: verify update public key with force idempotence
+  assert:
+    that:
+      - result is not changed
+
 - name: test delete ssh key
   hetzner.hcloud.ssh_key:
-    id: "{{ ssh_key.hcloud_ssh_key.id }}"
+    name: "{{ hcloud_ssh_key_name }}"
     state: absent
   register: result
 - name: verify absent ssh_key

tests/unit/module_utils/test_ssh.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+import pytest
+from ansible_collections.hetzner.hcloud.plugins.module_utils.ssh import (
+    ssh_public_key_md5_fingerprint,
+)
+
+
+@pytest.mark.parametrize(
+    ("public_key", "fingerprint"),
+    [
+        (
+            "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAILNWUEdTk1oxrjUZ5erbKUmJM3VxQ9DLocgt/HFohCf6 comment",
+            "ce:cf:37:b9:38:40:ad:80:b2:8b:2c:5c:83:b5:af:0b",
+        ),
+        (
+            "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAILNWUEdTk1oxrjUZ5erbKUmJM3VxQ9DLocgt/HFohCf6",  # No comment
+            "ce:cf:37:b9:38:40:ad:80:b2:8b:2c:5c:83:b5:af:0b",
+        ),
+        (
+            "ecdsa-sha2-nistp521 AAAAE2VjZHNhLXNoYTItbmlzdHA1MjEAAAAIbmlzdHA1MjEAAACFBABOUmmgxbhhauMg97GMwHcjWXM35MwFmlSKx7klWpPr3jMbabGQzINFVXexgf6Tru0D5a7NU/Hkx9t2yOtqKHJOIQB5/NKktqYelul4X56WYV/64RSm6xIjcolNao9fUbawnIwh9mvaQQg5v1BiJfPJ6p6LcWPunzfm6DztU1tHwLtjTw== comment",  # noqa: E501
+            "bf:61:7b:7f:ab:c7:af:25:aa:d7:d5:e8:5f:87:5c:66",
+        ),
+    ],
+)
+def test_ssh_public_key_md5_fingerprint(public_key: str, fingerprint: str):
+    assert ssh_public_key_md5_fingerprint(public_key) == fingerprint