Feature/content signing (#30)

Feature/content signing

Author: uilianries

bintray/bintray.py

@@ -514,3 +514,99 @@ def get_oss_licenses(self):
         """
         url = "{}/licenses/oss_licenses".format(Bintray.BINTRAY_URL)
         return self._requester.get(url)
+
+    # Content Signing
+
+    def get_org_gpg_public_key(self, org):
+        """ Get the organization GPG public key.
+
+            The response Content-Type format is 'application/pgp-keys'.
+
+        :param org: Organization name
+        :return: response Content-Type format as 'application/pgp-keys'.
+        """
+        url = "{}/orgs/{}/keys/gpg/public.key".format(Bintray.BINTRAY_URL, org)
+        return self._requester.get(url)
+
+    def get_user_gpg_public_key(self, user):
+        """ Get the subject GPG public key.
+
+            The response Content-Type format is 'application/pgp-keys'.
+
+        :param org: Organization name
+        :return: response Content-Type format as 'application/pgp-keys'.
+        """
+        url = "{}/users/{}/keys/gpg/public.key".format(Bintray.BINTRAY_URL, user)
+        return self._requester.get(url)
+
+    def gpg_sign_version(self, subject, repo, package, version, key_subject=None, passphrase=None,
+                         key_path=None):
+        """ GPG sign all files associated with the specified version.
+
+            GPG signing information may be needed
+
+            Security: Authenticated user with 'publish' permission.
+
+        :param subject: username or organization
+        :param repo: repository name
+        :param package: package name
+        :param version: package version
+        :param key_subject: Alternative Bintray subject for the GPG public key
+        :param passphrase: Optional private key passphrase, if required
+        :param key_path: Optional private key, if not stored in Bintray
+        :return: request response
+        """
+        url = "{}/gpg/{}/{}/{}/versions/{}".format(Bintray.BINTRAY_URL, subject, repo, package,
+                                                   version)
+        body = {}
+        if subject:
+            body['subject'] = key_subject
+        if passphrase:
+            body['passphrase'] = passphrase
+        if key_path:
+            with open(key_path, 'r') as fd:
+                body['private_key'] = fd.read()
+        body = None if body == {} else body
+        headers = None
+        if "passphrase" in body and len(body.keys()) == 1:
+            headers = {"X-GPG-PASSPHRASE": passphrase}
+            body = None
+
+        response = self._requester.post(url, json=body, headers=headers)
+
+        self._logger.info("Sign successfully: {}".format(url))
+        return response
+
+    def gpg_sign_file(self, subject, repo, file_path, key_subject=None, passphrase=None,
+                      key_path=None):
+        """ GPG sign the specified repository file.
+
+            GPG signing information may be needed
+
+        :param subject: username or organization
+        :param repo: repository name
+        :param file_path: file path to be signed
+        :param key_subject: Alternative Bintray subject for the GPG public key
+        :param passphrase: Optional private key passphrase, if required
+        :param key_path: Optional private key, if not stored in Bintray
+        :return: request response
+        """
+        url = "{}/gpg/{}/{}/{}".format(Bintray.BINTRAY_URL, subject, repo, file_path)
+        body = {}
+        if subject:
+            body['subject'] = key_subject
+        if passphrase:
+            body['passphrase'] = passphrase
+        if key_path:
+            with open(key_path, 'r') as fd:
+                body['private_key'] = fd.read()
+        body = None if body == {} else body
+        headers = None
+        if "passphrase" in body and len(body.keys()) == 1:
+            headers = {"X-GPG-PASSPHRASE": passphrase}
+            body = None
+
+        response = self._requester.post(url, json=body, headers=headers)
+
+        self._logger.info("Sign successfully: {}".format(url))
+        return response

bintray/requester.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 
 import requests
+import json
 
 from requests.auth import HTTPBasicAuth
 
@@ -31,7 +32,10 @@ def _add_status_code(self, response):
         :param response: Requests response
         :return: Response JSON
         """
-        json_data = response.json()
+        try:
+            json_data = response.json()
+        except json.decoder.JSONDecodeError:
+            json_data = {'message': response.content.decode()}
         if isinstance(json_data, list):
             json_data.append({"statusCode": response.status_code, "error": not response.ok})
         else:

tests/test_content_signing.py

@@ -0,0 +1,34 @@
+# -*- coding: utf-8 -*-
+
+import os
+from bintray.bintray import Bintray
+
+
+def test_get_org_gpg_public_key():
+    bintray = Bintray()
+    response = bintray.get_org_gpg_public_key("jfrog")
+    assert response.get("error") == False
+    assert response.get("statusCode") == 200
+    assert "BEGIN PGP PUBLIC KEY BLOCK" in response.get("message")
+
+
+def test_get_user_gpg_public_key():
+    bintray = Bintray()
+    response = bintray.get_user_gpg_public_key("bintray")
+    assert response.get("error") == False
+    assert response.get("statusCode") == 200
+    assert "BEGIN PGP PUBLIC KEY BLOCK" in response.get("message")
+
+
+def test_gpg_sign_version():
+    bintray = Bintray()
+    response = bintray.gpg_sign_version("uilianries", "generic", "statistics", "test")
+    assert {'error': False, 'message': 'success', 'statusCode': 200} == response or \
+           {'error': False, 'message': 'success', 'statusCode': 201} == response
+
+
+def test_gpg_sign_file():
+    bintray = Bintray()
+    response = bintray.gpg_sign_file("uilianries", "generic", "packages.json")
+    assert {'error': False, 'message': 'success', 'statusCode': 200} == response or \
+           {'error': False, 'message': 'success', 'statusCode': 201} == response