launchpad-reviewers team mailing list archive

[Tom Wardill <tom.wardill@xxxxxxxxxxxxx>]

Tom Wardill has proposed merging ~twom/launchpad:oci-store-username-unencrypted into launchpad:master.

Commit message:
Store username unencrypted if we have it

Requested reviews:
  Launchpad code reviewers (launchpad-reviewers)

For more details, see:
https://code.launchpad.net/~twom/launchpad/+git/launchpad/+merge/382674
-- 
Your team Launchpad code reviewers is requested to review the proposed merge of ~twom/launchpad:oci-store-username-unencrypted into launchpad:master.

diff --git a/lib/lp/oci/model/ociregistrycredentials.py b/lib/lp/oci/model/ociregistrycredentials.py
index 03b0f72..a82913f 100644
--- a/lib/lp/oci/model/ociregistrycredentials.py
+++ b/lib/lp/oci/model/ociregistrycredentials.py
@@ -77,11 +77,23 @@ class OCIRegistryCredentials(Storm):
         self.url = url
         self.setCredentials(credentials)
 
+    """
+    Conveniently, OCIRegistryCredentials._credentials is a dict, so we can adjust this a little without too much trouble.  Maybe it could end up being something like:
+
+    {
+        "username": [unencrypted username],
+        "credentials_encrypted": [encrypted container of everything else],
+    }
+    """
+
     def getCredentials(self):
         container = getUtility(IEncryptedContainer, "oci-registry-secrets")
         try:
-            return json.loads(container.decrypt(
+            data = json.loads(container.decrypt(
                 self._credentials['credentials_encrypted']).decode("UTF-8"))
+            if self._credentials.get("username"):
+                data["username"] = self._credentials["username"]
+            return data
         except CryptoError as e:
             # XXX twom 2020-03-18 This needs a better error
             # see SnapStoreClient.UnauthorizedUploadResponse
@@ -90,9 +102,18 @@ class OCIRegistryCredentials(Storm):
 
     def setCredentials(self, value):
         container = getUtility(IEncryptedContainer, "oci-registry-secrets")
-        self._credentials = {
-            "credentials_encrypted": removeSecurityProxy(
-                container.encrypt(json.dumps(value).encode('UTF-8')))}
+        if value.get("username"):
+            copy = value.copy()
+            username = copy["username"]
+            del copy["username"]
+            self._credentials = {
+                "username": username,
+                "credentials_encrypted": removeSecurityProxy(
+                    container.encrypt(json.dumps(copy).encode('UTF-8')))}
+        else:
+            self._credentials = {
+                "credentials_encrypted": removeSecurityProxy(
+                    container.encrypt(json.dumps(value).encode('UTF-8')))}
 
     def destroySelf(self):
         """See `IOCIRegistryCredentials`."""
diff --git a/lib/lp/oci/tests/test_ociregistrycredentials.py b/lib/lp/oci/tests/test_ociregistrycredentials.py
index 50953da..e0e7bcb 100644
--- a/lib/lp/oci/tests/test_ociregistrycredentials.py
+++ b/lib/lp/oci/tests/test_ociregistrycredentials.py
@@ -63,11 +63,54 @@ class TestOCIRegistryCredentials(OCIConfigHelperMixin, TestCaseWithFactory):
                 credentials=credentials))
         container = getUtility(IEncryptedContainer, "oci-registry-secrets")
         self.assertThat(oci_credentials._credentials, MatchesDict({
+            "username": Equals("foo"),
             "credentials_encrypted": AfterPreprocessing(
                 lambda value: json.loads(container.decrypt(value)),
-                Equals(credentials)),
+                Equals({"password": "bar"})),
             }))
 
+    def test_credentials_set_empty(self):
+        owner = self.factory.makePerson()
+        oci_credentials = self.factory.makeOCIRegistryCredentials(
+            owner=owner,
+            url='http://example.org',
+            credentials={})
+        with person_logged_in(owner):
+            self.assertThat(oci_credentials.getCredentials(), MatchesDict({}))
+
+    def test_credentials_set_no_password(self):
+        owner = self.factory.makePerson()
+        oci_credentials = self.factory.makeOCIRegistryCredentials(
+            owner=owner,
+            url='http://example.org',
+            credentials={"username": "test"})
+        with person_logged_in(owner):
+            self.assertThat(oci_credentials.getCredentials(), MatchesDict({
+                "username": Equals("test")}))
+
+    def test_credentials_set_no_username(self):
+        owner = self.factory.makePerson()
+        oci_credentials = self.factory.makeOCIRegistryCredentials(
+            owner=owner,
+            url='http://example.org',
+            credentials={"password": "test"})
+        with person_logged_in(owner):
+            self.assertThat(oci_credentials.getCredentials(), MatchesDict({
+                "password": Equals("test")}))
+
+    def test_credentials_set_encrypts_other_data(self):
+        owner = self.factory.makePerson()
+        oci_credentials = self.factory.makeOCIRegistryCredentials(
+            owner=owner,
+            url='http://example.org',
+            credentials={
+                "username": "foo", "password": "bar", "other": "baz"})
+        with person_logged_in(owner):
+            self.assertThat(oci_credentials.getCredentials(), MatchesDict({
+                "username": Equals("foo"),
+                "password": Equals("bar"),
+                "other": Equals("baz")}))
+
 
 class TestOCIRegistryCredentialsSet(OCIConfigHelperMixin, TestCaseWithFactory):