condensed two main execution paths and refactored functions

plugins/modules/github_gpg_key.py

@@ -33,34 +33,30 @@ options:
   name:
     description:
       - GPG key name
-    required: true
     type: str
   armored_public_key:
     description:
       - ASCII-armored GPG public key value. Required when O(state=present).
     type: str
+  gpg_key_id:
+    description:
+      - GPG key id. Required when O(state=absent).
+    type: int
   state:
     description:
       - Whether to remove a key, ensure that it exists, or update its value.
     choices: [ 'present', 'absent' ]
     default: 'present'
     type: str
-  force:
-    description:
-      - The default is V(true), which will replace the existing remote key
-        if it is different than O(armored_public_key). If V(false), the key will only be
-        set if no key with the given O(name) exists.
-    type: bool
-    default: true
 '''
 
 RETURN = '''
-deleted_keys:
+deleted_key:
-    description: An array of key objects that were deleted. Only present on state=absent
+    description: A GPG key that was deleted from GitHub. Only present on O(state=absent).
     type: list
     elements: dict
-    returned: When O(state=absent)
+    returned: changed or success 
-    sample: [{
+    sample: {
         'id': 3,
         'name': "Octocat's GPG Key",
         'primary_key_id': 2,
@@ -92,49 +88,48 @@ deleted_keys:
         'expires_at': '2016-03-24T11:31:04-07:00',
         'revoked': false,
         'raw_key': 'string'
-    }]
+    }
-matching_keys:
+matching_key:
-    description: An array of keys matching the specified name. Only present on state=present
+    description: A matching GPG key found on GitHub. Only present when O(state=present) and no new key is created.
-    type: list
-    elements: dict
-    returned: When O(state=present)
-    sample: [{
-        'id': 3,
-        'name': "Octocat's GPG Key",
-        'primary_key_id': 2,
-        'key_id': '3262EFF25BA0D270',
-        'public_key': 'xsBNBFayYZ...',
-        'emails': [{
-            'email': 'octocat@users.noreply.github.com',
-            'verified': true
-        }],
-        'subkeys': [{
-            'id': 4,
-            'primary_key_id': 3,
-            'key_id': '4A595D4C72EE49C7',
-            'public_key': 'zsBNBFayYZ...',
-            'emails': [],
-            'can_sign': false,
-            'can_encrypt_comms': true,
-            'can_encrypt_storage': true,
-            'can_certify': false,
-            'created_at': '2016-03-24T11:31:04-06:00',
-            'expires_at': '2016-03-24T11:31:04-07:00',
-            'revoked': false
-        }],
-        'can_sign': true,
-        'can_encrypt_comms': false,
-        'can_encrypt_storage': false,
-        'can_certify': true,
-        'created_at': '2016-03-24T11:31:04-06:00',
-        'expires_at': '2016-03-24T11:31:04-07:00',
-        'revoked': false,
-        'raw_key': 'string'
-    }]
-key:
-    description: Metadata about the key just created. Only present on state=present
     type: dict
-    returned: When O(state=present)
+    returned: not changed
+    sample: {
+        'id': 3,
+        'name': "Octocat's GPG Key",
+        'primary_key_id': 2,
+        'key_id': '3262EFF25BA0D270',
+        'public_key': 'xsBNBFayYZ...',
+        'emails': [{
+            'email': 'octocat@users.noreply.github.com',
+            'verified': true
+        }],
+        'subkeys': [{
+            'id': 4,
+            'primary_key_id': 3,
+            'key_id': '4A595D4C72EE49C7',
+            'public_key': 'zsBNBFayYZ...',
+            'emails': [],
+            'can_sign': false,
+            'can_encrypt_comms': true,
+            'can_encrypt_storage': true,
+            'can_certify': false,
+            'created_at': '2016-03-24T11:31:04-06:00',
+            'expires_at': '2016-03-24T11:31:04-07:00',
+            'revoked': false
+        }],
+        'can_sign': true,
+        'can_encrypt_comms': false,
+        'can_encrypt_storage': false,
+        'can_certify': true,
+        'created_at': '2016-03-24T11:31:04-06:00',
+        'expires_at': '2016-03-24T11:31:04-07:00',
+        'revoked': false,
+        'raw_key': 'string'
+    }
+new_key:
+    description: A new GPG key that was added to GitHub. Only present on O(state=present).
+    type: dict
+    returned: changed or success
     sample: {
         'id': 3,
         'name': "Octocat's GPG Key",
@@ -171,139 +166,127 @@ key:
 '''
 
 EXAMPLES = '''
-- name: Authorize key with GitHub
+- name: Add GitHub GPG key
   community.general.github_gpg_key:
-    name: My GPG Key
+    state: present
     token: '{{ token }}'
+    name: My GPG Key
     armored_public_key: '{{ armored_public_key }}'
+
+- name: Delete GitHub GPG key
+  community.general.github_gpg_key:
+    state: absent
+    token: '{{ token }}'
+    gpg_key_id: '{{ gpg_key_id }}'
 '''
 
 import json
-import re
-
 from ansible.module_utils.basic import AnsibleModule
-from ansible.module_utils.urls import fetch_url
+from ansible.module_utils.urls import open_url
-
-from ansible_collections.community.general.plugins.module_utils.datetime import now
 
 
-API_BASE = 'https://api.github.com/user/gpg_keys'
+GITHUB_GPG_REST_API_URL = 'https://api.github.com/user/gpg_keys'
 
 
-class GitHubResponse(object):
+def ensure_gpg_key_absent(headers, gpg_key_id, check_mode):
-    def __init__(self, response, info):
+    changed = False
-        self.content = response.read()
+    deleted_key = {}
-        self.info = info
 
-    def json(self):
+    method = 'GET' if check_mode else 'DELETE'
-        return json.loads(self.content)
+    response = open_url(
-
+        url=GITHUB_GPG_REST_API_URL+'/'+gpg_key_id,
-    def links(self):
+        method=method,
-        links = {}
+        headers=headers
-        if 'link' in self.info:
+    )
-            link_header = self.info['link']
+    if response.status == 200:
-            matches = re.findall('<([^>]+)>; rel="([^"]+)"', link_header)
+        changed = True
-            for url, rel in matches:
+        deleted_key = json.loads(response.read())
-                links[rel] = url
-        return links
-
-
-class GitHubSession(object):
-    def __init__(self, module, token):
-        self.module = module
-        self.token = token
-
-    def request(self, method, url, data=None):
-        headers = {
-            'Authorization': 'token {}'.format(self.token),
-            'Content-Type': 'application/json',
-            'Accept': 'application/vnd.github.v3+json'
-        }
-        response, info = fetch_url(
-            self.module, url, method=method, data=data, headers=headers)
-        if not (200 <= info['status'] < 400):
-            self.module.fail_json('failed to send request {0} to {1}: {2}'.format(method, url, info['msg']))
-        return GitHubResponse(response, info)
-
-
-def get_all_keys(session):
-    url = API_BASE
-    result = []
-    while url:
-        r = session.request('GET', url)
-        result.extend(r.json())
-        url = r.links().get('next')
-    return result
-
-
-def create_key(session, name, armored_public_key, check_mode):
-    if check_mode:
-        now_t = now()
-        return {}
-    else:
-        return session.request(
-            'POST',
-            API_BASE,
-            data=json.dumps({'name': name, 'raw_key': armored_public_key})).json()
-
-
-def delete_keys(session, to_delete, check_mode):
-    if check_mode:
-        return
-
-    for key in to_delete:
-        session.request('DELETE', API_BASE + '/' + key['id'])
-
-
-def ensure_key_absent(session, name, check_mode):
-    to_delete = [key for key in get_all_keys(session) if key['name'] == name]
-    delete_keys(session, to_delete, check_mode=check_mode)
-
-    return {'changed': bool(to_delete),
-            'deleted_keys': to_delete}
-
-
-def ensure_key_present(module, session, name, armored_public_key, force, check_mode):
-    all_keys = get_all_keys(session)
-    matching_keys = [k for k in all_keys if k['name'] == name]
-    deleted_keys = []
-
-    new_signature = armored_public_key
-    for key in all_keys:
-        existing_signature = key['raw_key']
-        if new_signature == existing_signature and key['name'] != name:
-            module.fail_json('Another key with the same content is already registered under the name |{0}|'.format(key['title']))
-
-    if matching_keys and force and matching_keys[0]['raw_key'] != new_signature:
-        delete_keys(session, matching_keys, check_mode=check_mode)
-        deleted_keys, matching_keys = matching_keys, []
-
-    if not matching_keys:
-        key = create_key(session, name, armored_public_key, check_mode=check_mode)
-    else:
-        key = matching_keys[0]
 
     return {
-        'changed': bool(deleted_keys or not matching_keys),
+        'changed': changed,
-        'deleted_keys': deleted_keys,
+        'deleted_key': deleted_key
-        'matching_keys': matching_keys,
-        'key': key
     }
 
 
-def validate_key(module, armored_public_key):
+def ensure_gpg_key_present(headers, name, armored_public_key, check_mode):
+    changed = False
+    matching_key = {}
+    new_key = {}
+
     armored_public_key_parts = armored_public_key.splitlines()
-    if armored_public_key_parts[0] != '-----BEGIN PGP PUBLIC KEY BLOCK-----' or armored_public_key_parts[-1] != '-----END PGP PUBLIC KEY BLOCK-----':
+    if (armored_public_key_parts[0] != '-----BEGIN PGP PUBLIC KEY BLOCK-----') \
-        module.fail_json(msg='"armored_public_key" parameter has an invalid format')
+        or (armored_public_key_parts[-1] != '-----END PGP PUBLIC KEY BLOCK-----'):
+        raise Exception('GPG key must have ASCII armor')
 
+    response = open_url(
+        url=GITHUB_GPG_REST_API_URL,
+        method='GET',
+        headers=headers
+    )
+    if response.status != 200:
+        raise Exception(
+            "Failed to check for matching GPG key: {} {}"
+            .format(response.status, response.reason)
+        )
 
-def run_module(module, params, check_mode):
+    keys = json.loads(response.read())
-    session = GitHubSession(module, params['token'])
+    for key in keys:
+        if key['raw_key'] == armored_public_key:
+            matching_key = key
+            break
+
+    if not matching_key:
+        response = open_url(
+            url=GITHUB_GPG_REST_API_URL,
+            method='POST',
+            data={'name': name, 'armored_public_key': armored_public_key},
+            headers=headers
+        )
+        if response.status != 201:
+            raise Exception(
+                "Failed to create GPG key: {} {}"
+                .format(response.status, response.reason)
+            )
+
+        changed = True
+        new_key = json.loads(response.json())
+        if check_mode and new_key:
+            response = open_url(
+                url=GITHUB_GPG_REST_API_URL+'/'+new_key['key_id'],
+                method='DELETE',
+                headers=headers
+            )
+            if response.status != 200:
+                raise Exception(
+                    "Failed to undo changes (check_mode=true): {} {}"
+                    .format(response.status, response.reason)
+                )
+
+    return {
+        'changed': changed,
+        'matching_key': matching_key,
+        'new_key': new_key 
+    }
+
+def run_module(params, check_mode):
+    headers = {
+        'Accept': 'application/vnd.github+json',
+        'Authorization': 'Bearer {}'.format(params['token']),
+        'X-GitHub-Api-Version': '2022-11-28',
+    }
     if params['state'] == 'present':
-        validate_key(module, params['armored_public_key'])
+        result = ensure_gpg_key_present(
-        result = ensure_key_present(module, session, params['name'], params['armored_public_key'], params['force'], check_mode)
+            headers,
+            params['name'],
+            params['armored_public_key'],
+            check_mode
+        )
     else:
-        result = ensure_key_absent(session, params['name'], check_mode)
+        result = ensure_gpg_key_absent(
+            headers,
+            params['gpg_key_id'],
+            check_mode
+        )
     return result
 
 
@@ -312,13 +295,14 @@ def main():
         argument_spec=dict(
             state=dict(type='str', default='present', choices=['present', 'absent']),
             token=dict(type='str', required=True, no_log=True),
-            name=dict(type='str', required=True),
+            name=dict(type='str', no_log=True),
             armored_public_key=dict(type='str', no_log=True),
-            force=dict(type='bool', default=True)
+            gpg_key_id=dict(type='int', no_log=True)
         ),
         supports_check_mode=True,
         required_if=[
             ['state', 'present', ['armored_public_key']],
+            ['state', 'absent', ['gpg_key_id']],
         ]
     )