invoke all base64 ops via wallet utils

aries_cloudagent/messaging/decorators/tests/test_attach_decorator.py

@@ -1,10 +1,11 @@
 from datetime import datetime, timezone
 from unittest import TestCase
 
-import base64
 import json
 import uuid
 
+from ....wallet.util import bytes_to_b64
+
 from ..attach_decorator import AttachDecorator, AttachDecoratorData
 
 
@@ -16,7 +17,7 @@ class TestAttachDecorator(TestCase):
     lastmod_time = datetime.now().replace(tzinfo=timezone.utc).isoformat(" ", "seconds")
     byte_count = 123456
     data_b64 = AttachDecoratorData(
-        base64_=base64.b64encode('sample image with padding'.encode()).decode()
+        base64_=bytes_to_b64(b'sample image with padding')
     )
     data_json = AttachDecoratorData(
         json_=json.dumps({'preference': 'hasselback', 'variety': 'russet'})

aries_cloudagent/messaging/issue_credential/v1_0/messages/inner/credential_preview.py

@@ -3,10 +3,9 @@
 
 from typing import Sequence
 
-import base64
-
 from marshmallow import fields, validate
 
+from ......wallet.util import b64_to_str
 from .....models.base import BaseModel, BaseModelSchema
 from .....util import canon
 from ...message_types import CREDENTIAL_PREVIEW
@@ -61,8 +60,7 @@ def list_plain(plain: dict):
     def b64_decoded_value(self) -> str:
         """Value, base64-decoded if applicable."""
 
-        return base64.b64decode(self.value.encode()).decode(
-        ) if self.value and self.mime_type else self.value
+        return b64_to_str(self.value) if self.value and self.mime_type else self.value
 
     def __eq__(self, other):
         """Equality comparator."""
@@ -153,14 +151,14 @@ def attr_dict(self, decode: bool = False):
         """
 
         return {
-            attr.name: base64.b64decode(attr.value.encode()).decode()
-            if attr.mime_type and decode else attr.value
-            for attr in self.attributes
+            attr.name: b64_to_str(
+                attr.value
+            ) if attr.mime_type and decode else attr.value for attr in self.attributes
         }
 
     def mime_types(self):
         """
-        Return per-attribute mapping from name to MIME type and encoding.
+        Return per-attribute mapping from name to MIME type.
 
         Return empty dict if no attribute has MIME type.
 

aries_cloudagent/messaging/present_proof/v1_0/messages/inner/presentation_preview.py

@@ -6,11 +6,10 @@
 from time import time
 from typing import Mapping, Sequence
 
-import base64
-
 from marshmallow import fields, validate
 
 from ......ledger.indy import IndyLedger
+from ......wallet.util import b64_to_str
 from .....models.base import BaseModel, BaseModelSchema
 from .....util import canon
 from .....valid import INDY_CRED_DEF_ID, INDY_PREDICATE
@@ -120,7 +119,6 @@ def __init__(
             name: attribute name
             cred_def_id: credential definition identifier
                 (None for self-attested attribute)
-            encoding: encoding (omit or "base64")
             mime_type: MIME type
             value: attribute value as credential stores it
                 (None for unrevealed attribute)
@@ -172,8 +170,7 @@ def posture(self) -> "PresAttrSpec.Posture":
     def b64_decoded_value(self) -> str:
         """Value, base64-decoded if applicable."""
 
-        return base64.b64decode(self.value.encode()).decode(
-        ) if self.value and self.mime_type else self.value
+        return b64_to_str(self.value) if self.value and self.mime_type else self.value
 
     def satisfies(self, pred_spec: PresPredSpec):
         """Whether current specified attribute satisfied input specified predicate."""

aries_cloudagent/messaging/tests/test_utils.py → aries_cloudagent/messaging/tests/test_util.py

@@ -11,7 +11,7 @@
 )
 
 
-class TestUtils(TestCase):
+class TestUtil(TestCase):
     def test_parse(self):
         now = datetime_now()
         assert isinstance(now, datetime)

aries_cloudagent/wallet/tests/test_util.py

@@ -0,0 +1,45 @@
+from datetime import datetime, timezone
+from unittest import mock, TestCase
+
+from ..util import (
+    b64_to_bytes,
+    b64_to_str,
+    bytes_to_b64,
+    str_to_b64,
+    set_urlsafe_b64,
+    b58_to_bytes,
+    bytes_to_b58,
+)
+
+
+BYTES = b"\xe0\xa0\xbe"  # chr(2110).encode(), least with + in b64 encoding
+STR = "Hello World"
+
+
+class TestUtil(TestCase):
+
+    def test_b64_urlsafe(self):
+        for urlsafe in (False, True):
+            CHAR62 = ['+', '-']
+            b64 = bytes_to_b64(BYTES, urlsafe=urlsafe)
+            assert CHAR62[urlsafe] in b64
+
+            b64 = set_urlsafe_b64(b64, urlsafe=urlsafe)
+            assert CHAR62[urlsafe] in b64
+            assert CHAR62[not urlsafe] not in b64
+
+            b64 = set_urlsafe_b64(b64, urlsafe=(not urlsafe))
+            assert CHAR62[urlsafe] not in b64
+            assert CHAR62[not urlsafe] in b64
+
+    def test_b64_str(self):
+        b64 = str_to_b64(STR)
+        assert b64_to_str(b64) == STR
+
+        b64 = str_to_b64(STR, urlsafe=True)
+        assert b64_to_str(b64, urlsafe=True) == STR
+        assert b64_to_str(b64, encoding="ascii") == STR
+
+    def test_b58(self):
+        b58 = bytes_to_b58(BYTES)
+        assert b58_to_bytes(b58) == BYTES

aries_cloudagent/wallet/util.py

@@ -11,13 +11,23 @@ def b64_to_bytes(val: str, urlsafe=False) -> bytes:
     return base64.b64decode(val)
 
 
+def b64_to_str(val: str, urlsafe=False, encoding=None) -> str:
+    """Convert a base 64 string to string on input encoding (default utf-8)."""
+    return b64_to_bytes(val, urlsafe).decode(encoding or "utf-8")
+
+
 def bytes_to_b64(val: bytes, urlsafe=False) -> str:
     """Convert a byte string to base 64."""
     if urlsafe:
         return base64.urlsafe_b64encode(val).decode("ascii")
     return base64.b64encode(val).decode("ascii")
 
 
+def str_to_b64(val: str, urlsafe=False, encoding=None) -> str:
+    """Convert a string to base64 string on input encoding (default utf-8)."""
+    return bytes_to_b64(val.encode(encoding or "utf-8"), urlsafe)
+
+
 def set_urlsafe_b64(val: str, urlsafe: bool = True) -> str:
     """Set URL safety in base64 encoding."""
     if urlsafe: