From 481b51fc69e604a2d510e391f9d610603ef78a85 Mon Sep 17 00:00:00 2001
From: Tom Prince <tom.prince@private.storage>
Date: Mon, 6 Dec 2021 11:55:43 -0700
Subject: [PATCH] Convert `Pass`, `UnblinedToken`, `RandomToken` and `Voucher`
to use bytes internally.
---
src/_zkapauthorizer/_storage_client.py | 2 +-
src/_zkapauthorizer/controller.py | 39 ++++-----
src/_zkapauthorizer/model.py | 80 +++++++++++--------
src/_zkapauthorizer/resource.py | 4 +-
src/_zkapauthorizer/tests/privacypass.py | 12 +--
src/_zkapauthorizer/tests/storage_common.py | 2 +-
src/_zkapauthorizer/tests/strategies.py | 43 +++-------
.../tests/test_client_resource.py | 9 ++-
src/_zkapauthorizer/tests/test_controller.py | 8 +-
.../tests/test_storage_protocol.py | 2 +-
.../tests/test_storage_server.py | 4 +-
src/_zkapauthorizer/validators.py | 9 ++-
12 files changed, 103 insertions(+), 111 deletions(-)
diff --git a/src/_zkapauthorizer/_storage_client.py b/src/_zkapauthorizer/_storage_client.py
index 622299f..5f4e568 100644
--- a/src/_zkapauthorizer/_storage_client.py
+++ b/src/_zkapauthorizer/_storage_client.py
@@ -215,7 +215,7 @@ def _encode_passes(group):
:return list[bytes]: The encoded form of the passes in the given group.
"""
- return list(t.pass_text.encode("ascii") for t in group.passes)
+ return list(t.pass_bytes for t in group.passes)
@implementer(IStorageServer)
diff --git a/src/_zkapauthorizer/controller.py b/src/_zkapauthorizer/controller.py
index edda44e..ab52c5b 100644
--- a/src/_zkapauthorizer/controller.py
+++ b/src/_zkapauthorizer/controller.py
@@ -318,7 +318,7 @@ class RecordingRedeemer(object):
def dummy_random_tokens(voucher, counter, count):
- v = urlsafe_b64decode(voucher.number.encode("ascii"))
+ v = urlsafe_b64decode(voucher.number)
def dummy_random_token(n):
return RandomToken(
@@ -326,7 +326,7 @@ def dummy_random_tokens(voucher, counter, count):
# length) - 4 (fixed-width counter)
b64encode(
v + u"{:0>4}{:0>60}".format(counter, n).encode("ascii"),
- ).decode("ascii"),
+ ),
)
return list(dummy_random_token(n) for n in range(count))
@@ -380,9 +380,9 @@ class DummyRedeemer(object):
)
def dummy_unblinded_token(random_token):
- random_value = b64decode(random_token.token_value.encode("ascii"))
+ random_value = b64decode(random_token.token_value)
unblinded_value = random_value + b"x" * (96 - len(random_value))
- return UnblindedToken(b64encode(unblinded_value).decode("ascii"))
+ return UnblindedToken(b64encode(unblinded_value))
return succeed(
RedemptionResult(
@@ -397,17 +397,13 @@ class DummyRedeemer(object):
# can include in the resulting Pass. This ensures the pass values
# will be unique if and only if the unblinded tokens were unique
# (barring improbable hash collisions).
- token_digest = (
- sha256(token.unblinded_token.encode("ascii"))
- .hexdigest()
- .encode("ascii")
- )
+ token_digest = sha256(token.unblinded_token).hexdigest().encode("ascii")
preimage = b"preimage-" + token_digest[len(b"preimage-") :]
signature = b"signature-" + token_digest[len(b"signature-") :]
return Pass(
- b64encode(preimage).decode("ascii"),
- b64encode(signature).decode("ascii"),
+ b64encode(preimage),
+ b64encode(signature),
)
return list(token_to_pass(token) for token in unblinded_tokens)
@@ -489,9 +485,7 @@ class RistrettoRedeemer(object):
def random_tokens_for_voucher(self, voucher, counter, count):
return list(
RandomToken(
- challenge_bypass_ristretto.RandomToken.create()
- .encode_base64()
- .decode("ascii"),
+ challenge_bypass_ristretto.RandomToken.create().encode_base64(),
)
for n in range(count)
)
@@ -499,9 +493,7 @@ class RistrettoRedeemer(object):
@inlineCallbacks
def redeemWithCounter(self, voucher, counter, encoded_random_tokens):
random_tokens = list(
- challenge_bypass_ristretto.RandomToken.decode_base64(
- token.token_value.encode("ascii")
- )
+ challenge_bypass_ristretto.RandomToken.decode_base64(token.token_value)
for token in encoded_random_tokens
)
blinded_tokens = list(token.blind() for token in random_tokens)
@@ -509,7 +501,7 @@ class RistrettoRedeemer(object):
self._api_root.child(u"v1", u"redeem").to_text(),
dumps(
{
- u"redeemVoucher": voucher.number,
+ u"redeemVoucher": voucher.number.decode("ascii"),
u"redeemCounter": counter,
u"redeemTokens": list(
token.encode_base64() for token in blinded_tokens
@@ -570,8 +562,7 @@ class RistrettoRedeemer(object):
)
self._log.info("Validated proof")
unblinded_tokens = list(
- UnblindedToken(token.encode_base64().decode("ascii"))
- for token in clients_unblinded_tokens
+ UnblindedToken(token.encode_base64()) for token in clients_unblinded_tokens
)
returnValue(
RedemptionResult(
@@ -586,7 +577,7 @@ class RistrettoRedeemer(object):
assert all(isinstance(element, UnblindedToken) for element in unblinded_tokens)
unblinded_tokens = list(
challenge_bypass_ristretto.UnblindedToken.decode_base64(
- token.unblinded_token.encode("ascii")
+ token.unblinded_token
)
for token in unblinded_tokens
)
@@ -600,8 +591,8 @@ class RistrettoRedeemer(object):
clients_preimages = list(token.preimage() for token in unblinded_tokens)
passes = list(
Pass(
- preimage.encode_base64().decode("ascii"),
- signature.encode_base64().decode("ascii"),
+ preimage.encode_base64(),
+ signature.encode_base64(),
)
for (preimage, signature) in zip(clients_preimages, clients_signatures)
)
@@ -805,7 +796,7 @@ class PaymentController(object):
@inlineCallbacks
def redeem(self, voucher, num_tokens=None):
"""
- :param unicode voucher: A voucher to redeem.
+ :param bytes voucher: A voucher to redeem.
:param int num_tokens: A number of tokens to redeem.
"""
diff --git a/src/_zkapauthorizer/model.py b/src/_zkapauthorizer/model.py
index 27e2fe9..594afc4 100644
--- a/src/_zkapauthorizer/model.py
+++ b/src/_zkapauthorizer/model.py
@@ -247,7 +247,7 @@ class VoucherStore(object):
@with_cursor
def get(self, cursor, voucher):
"""
- :param unicode voucher: The text value of a voucher to retrieve.
+ :param bytes voucher: The text value of a voucher to retrieve.
:return Voucher: The voucher object that matches the given value.
"""
@@ -260,7 +260,7 @@ class VoucherStore(object):
WHERE
[number] = ?
""",
- (voucher,),
+ (voucher.decode("ascii"),),
)
refs = cursor.fetchall()
if len(refs) == 0:
@@ -274,7 +274,7 @@ class VoucherStore(object):
existing) to the database. If the (voucher, counter) pair is already
present, do nothing.
- :param unicode voucher: The text value of a voucher with which to
+ :param bytes voucher: The text value of a voucher with which to
associate the tokens.
:param int expected_tokens: The total number of tokens for which this
@@ -302,7 +302,7 @@ class VoucherStore(object):
FROM [tokens]
WHERE [voucher] = ? AND [counter] = ?
""",
- (voucher, counter),
+ (voucher.decode("ascii"), counter),
)
rows = cursor.fetchall()
if len(rows) > 0:
@@ -312,26 +312,35 @@ class VoucherStore(object):
voucher=voucher,
counter=counter,
)
- tokens = list(RandomToken(token_value) for (token_value,) in rows)
+ tokens = list(
+ RandomToken(token_value.encode("ascii")) for (token_value,) in rows
+ )
else:
tokens = get_tokens()
self._log.info(
"Persisting {count} random tokens for a voucher ({voucher}[{counter}]).",
count=len(tokens),
- voucher=voucher,
+ voucher=voucher.decode("ascii"),
counter=counter,
)
cursor.execute(
"""
INSERT OR IGNORE INTO [vouchers] ([number], [expected-tokens], [created]) VALUES (?, ?, ?)
""",
- (voucher, expected_tokens, self.now()),
+ (voucher.decode("ascii"), expected_tokens, self.now()),
)
cursor.executemany(
"""
INSERT INTO [tokens] ([voucher], [counter], [text]) VALUES (?, ?, ?)
""",
- list((voucher, counter, token.token_value) for token in tokens),
+ list(
+ (
+ voucher.decode("ascii"),
+ counter,
+ token.token_value.decode("ascii"),
+ )
+ for token in tokens
+ ),
)
return tokens
@@ -385,7 +394,7 @@ class VoucherStore(object):
"""
Store some unblinded tokens received from redemption of a voucher.
- :param unicode voucher: The voucher associated with the unblinded
+ :param bytes voucher: The voucher associated with the unblinded
tokens. This voucher will be marked as redeemed to indicate it
has fulfilled its purpose and has no further use for us.
@@ -417,7 +426,7 @@ class VoucherStore(object):
"""
INSERT INTO [redemption-groups] ([voucher], [public-key], [spendable]) VALUES (?, ?, ?)
""",
- (voucher, public_key, spendable),
+ (voucher.decode("ascii"), public_key, spendable),
)
group_id = cursor.lastrowid
@@ -443,7 +452,7 @@ class VoucherStore(object):
token_count_increase,
sequestered_count_increase,
self.now(),
- voucher,
+ voucher.decode("ascii"),
),
)
if cursor.rowcount == 0:
@@ -453,7 +462,7 @@ class VoucherStore(object):
self._insert_unblinded_tokens(
cursor,
- list(t.unblinded_token for t in unblinded_tokens),
+ list(t.unblinded_token.decode("ascii") for t in unblinded_tokens),
group_id,
)
@@ -471,7 +480,7 @@ class VoucherStore(object):
WHERE [number] = ?
AND [state] = "pending"
""",
- (self.now(), voucher),
+ (self.now(), voucher.decode("ascii")),
)
if cursor.rowcount == 0:
# Was there no matching voucher or was it in the wrong state?
@@ -481,7 +490,7 @@ class VoucherStore(object):
FROM [vouchers]
WHERE [number] = ?
""",
- (voucher,),
+ (voucher.decode("ascii"),),
)
rows = cursor.fetchall()
if len(rows) == 0:
@@ -541,7 +550,7 @@ class VoucherStore(object):
""",
texts,
)
- return list(UnblindedToken(t) for (t,) in texts)
+ return list(UnblindedToken(t.encode("ascii")) for (t,) in texts)
@with_cursor
def count_unblinded_tokens(self, cursor):
@@ -577,7 +586,9 @@ class VoucherStore(object):
"""
INSERT INTO [to-discard] VALUES (?)
""",
- list((token.unblinded_token,) for token in unblinded_tokens),
+ list(
+ (token.unblinded_token.decode("ascii"),) for token in unblinded_tokens
+ ),
)
cursor.execute(
"""
@@ -614,7 +625,10 @@ class VoucherStore(object):
"""
INSERT INTO [invalid-unblinded-tokens] VALUES (?, ?)
""",
- list((token.unblinded_token, reason) for token in unblinded_tokens),
+ list(
+ (token.unblinded_token.decode("ascii"), reason)
+ for token in unblinded_tokens
+ ),
)
cursor.execute(
"""
@@ -640,7 +654,9 @@ class VoucherStore(object):
"""
INSERT INTO [to-reset] VALUES (?)
""",
- list((token.unblinded_token,) for token in unblinded_tokens),
+ list(
+ (token.unblinded_token.decode("ascii"),) for token in unblinded_tokens
+ ),
)
cursor.execute(
"""
@@ -813,7 +829,7 @@ class UnblindedToken(object):
and can be used to construct a privacy-preserving pass which can be
exchanged for service.
- :ivar unicode unblinded_token: The base64 encoded serialized form of the
+ :ivar bytes unblinded_token: The base64 encoded serialized form of the
unblinded token. This can be used to reconstruct a
``challenge_bypass_ristretto.UnblindedToken`` using that class's
``decode_base64`` method.
@@ -821,7 +837,7 @@ class UnblindedToken(object):
unblinded_token = attr.ib(
validator=attr.validators.and_(
- attr.validators.instance_of(unicode),
+ attr.validators.instance_of(bytes),
is_base64_encoded(),
has_length(128),
),
@@ -833,7 +849,7 @@ class Pass(object):
"""
A ``Pass`` instance completely represents a single Zero-Knowledge Access Pass.
- :ivar unicode pass_text: The text value of the pass. This can be sent to
+ :ivar bytes pass_bytes: The text value of the pass. This can be sent to
a service provider one time to anonymously prove a prior voucher
redemption. If it is sent more than once the service provider may
choose to reject it and the anonymity property is compromised. Pass
@@ -843,7 +859,7 @@ class Pass(object):
preimage = attr.ib(
validator=attr.validators.and_(
- attr.validators.instance_of(unicode),
+ attr.validators.instance_of(bytes),
is_base64_encoded(),
has_length(88),
),
@@ -851,27 +867,27 @@ class Pass(object):
signature = attr.ib(
validator=attr.validators.and_(
- attr.validators.instance_of(unicode),
+ attr.validators.instance_of(bytes),
is_base64_encoded(),
has_length(88),
),
)
@property
- def pass_text(self):
- return u"{} {}".format(self.preimage, self.signature)
+ def pass_bytes(self):
+ return b" ".join((self.preimage, self.signature))
@attr.s(frozen=True)
class RandomToken(object):
"""
- :ivar unicode token_value: The base64-encoded representation of the random
+ :ivar bytes token_value: The base64-encoded representation of the random
token.
"""
token_value = attr.ib(
validator=attr.validators.and_(
- attr.validators.instance_of(unicode),
+ attr.validators.instance_of(bytes),
is_base64_encoded(),
has_length(128),
),
@@ -1014,7 +1030,7 @@ class Error(object):
@attr.s(frozen=True)
class Voucher(object):
"""
- :ivar unicode number: The text string which gives this voucher its
+ :ivar bytes number: The byte string which gives this voucher its
identity.
:ivar datetime created: The time at which this voucher was added to this
@@ -1032,7 +1048,7 @@ class Voucher(object):
number = attr.ib(
validator=attr.validators.and_(
- attr.validators.instance_of(unicode),
+ attr.validators.instance_of(bytes),
is_base64_encoded(urlsafe_b64decode),
has_length(44),
),
@@ -1085,7 +1101,7 @@ class Voucher(object):
number, created, expected_tokens, state = row[:4]
return cls(
- number=number,
+ number=number.encode("ascii"),
expected_tokens=expected_tokens,
# All Python datetime-based date/time libraries fail to handle
# leap seconds. This parse call might raise an exception of the
@@ -1135,7 +1151,7 @@ class Voucher(object):
raise ValueError("Unrecognized state {!r}".format(state_json))
return cls(
- number=values[u"number"],
+ number=values[u"number"].encode("ascii"),
expected_tokens=values[u"expected-tokens"],
created=None
if values[u"created"] is None
@@ -1152,7 +1168,7 @@ class Voucher(object):
def to_json_v1(self):
state = self.state.to_json_v1()
return {
- u"number": self.number,
+ u"number": self.number.decode("ascii"),
u"expected-tokens": self.expected_tokens,
u"created": None if self.created is None else self.created.isoformat(),
u"state": state,
diff --git a/src/_zkapauthorizer/resource.py b/src/_zkapauthorizer/resource.py
index 4f955e2..f5db053 100644
--- a/src/_zkapauthorizer/resource.py
+++ b/src/_zkapauthorizer/resource.py
@@ -415,7 +415,7 @@ class _VoucherCollection(Resource):
self._log.info(
"Accepting a voucher ({voucher}) for redemption.", voucher=voucher
)
- self._controller.redeem(voucher)
+ self._controller.redeem(voucher.encode("ascii"))
return b""
def render_GET(self, request):
@@ -434,7 +434,7 @@ class _VoucherCollection(Resource):
if not is_syntactic_voucher(voucher):
return bad_request()
try:
- voucher = self._store.get(voucher)
+ voucher = self._store.get(voucher.encode("ascii"))
except KeyError:
return NoResource()
return VoucherView(self._controller.incorporate_transient_state(voucher))
diff --git a/src/_zkapauthorizer/tests/privacypass.py b/src/_zkapauthorizer/tests/privacypass.py
index a3820b0..83c5494 100644
--- a/src/_zkapauthorizer/tests/privacypass.py
+++ b/src/_zkapauthorizer/tests/privacypass.py
@@ -34,7 +34,7 @@ def make_passes(signing_key, for_message, random_tokens):
:param list[challenge_bypass_ristretto.RandomToken] random_tokens: The
random tokens to feed in to the pass generation process.
- :return list[unicode]: The privacy passes. The returned list has one
+ :return list[Pass]: The privacy passes. The returned list has one
element for each element of ``random_tokens``.
"""
blinded_tokens = list(token.blind() for token in random_tokens)
@@ -64,10 +64,12 @@ def make_passes(signing_key, for_message, random_tokens):
for verification_key in verification_keys
)
passes = list(
- u"{} {}".format(
- preimage.encode_base64().decode("ascii"),
- signature.encode_base64().decode("ascii"),
- ).encode("ascii")
+ b" ".join(
+ (
+ preimage.encode_base64(),
+ signature.encode_base64(),
+ )
+ )
for (preimage, signature) in zip(preimages, message_signatures)
)
return passes
diff --git a/src/_zkapauthorizer/tests/storage_common.py b/src/_zkapauthorizer/tests/storage_common.py
index c28cff6..ab4f2be 100644
--- a/src/_zkapauthorizer/tests/storage_common.py
+++ b/src/_zkapauthorizer/tests/storage_common.py
@@ -164,7 +164,7 @@ def get_passes(message, count, signing_key):
and bound to the given message.
"""
return list(
- Pass(*pass_.split(u" "))
+ Pass(*pass_.split(b" "))
for pass_ in make_passes(
signing_key,
message,
diff --git a/src/_zkapauthorizer/tests/strategies.py b/src/_zkapauthorizer/tests/strategies.py
index dc8763a..c439874 100644
--- a/src/_zkapauthorizer/tests/strategies.py
+++ b/src/_zkapauthorizer/tests/strategies.py
@@ -548,19 +548,10 @@ def share_parameters():
def vouchers():
"""
- Build unicode strings in the format of vouchers.
+ Build byte strings in the format of vouchers.
"""
- return (
- binary(
- min_size=32,
- max_size=32,
- )
- .map(
- urlsafe_b64encode,
- )
- .map(
- lambda voucher: voucher.decode("ascii"),
- )
+ return binary(min_size=32, max_size=32,).map(
+ urlsafe_b64encode,
)
@@ -674,40 +665,32 @@ def random_tokens():
length=_TOKEN_LENGTH,
entropy=4,
)
- .map(
- b64encode,
- )
- .map(
- lambda token: RandomToken(token.decode("ascii")),
- )
+ .map(b64encode)
+ .map(RandomToken)
)
def token_preimages():
"""
- Build ``unicode`` strings representing base64-encoded token preimages.
+ Build ``bytes`` strings representing base64-encoded token preimages.
"""
return byte_strings(
label=b"token-preimage",
length=_TOKEN_PREIMAGE_LENGTH,
entropy=4,
- ).map(
- lambda bs: b64encode(bs).decode("ascii"),
- )
+ ).map(b64encode)
def verification_signatures():
"""
- Build ``unicode`` strings representing base64-encoded verification
+ Build ``bytes`` strings representing base64-encoded verification
signatures.
"""
return byte_strings(
label=b"verification-signature",
length=_VERIFICATION_SIGNATURE_LENGTH,
entropy=4,
- ).map(
- lambda bs: b64encode(bs).decode("ascii"),
- )
+ ).map(b64encode)
def zkaps():
@@ -733,12 +716,8 @@ def unblinded_tokens():
length=_UNBLINDED_TOKEN_LENGTH,
entropy=4,
)
- .map(
- b64encode,
- )
- .map(
- lambda zkap: UnblindedToken(zkap.decode("ascii")),
- )
+ .map(b64encode)
+ .map(UnblindedToken)
)
diff --git a/src/_zkapauthorizer/tests/test_client_resource.py b/src/_zkapauthorizer/tests/test_client_resource.py
index b4e928c..7e3e9ab 100644
--- a/src/_zkapauthorizer/tests/test_client_resource.py
+++ b/src/_zkapauthorizer/tests/test_client_resource.py
@@ -493,7 +493,8 @@ class UnblindedTokenTests(TestCase):
dumps(
{
u"unblinded-tokens": list(
- token.unblinded_token for token in unblinded_tokens
+ token.unblinded_token.decode("ascii")
+ for token in unblinded_tokens
)
}
)
@@ -941,7 +942,7 @@ class VoucherTests(TestCase):
)
root = root_from_config(config, datetime.now)
agent = RequestTraversalAgent(root)
- data = BytesIO(dumps({u"voucher": voucher}))
+ data = BytesIO(dumps({u"voucher": voucher.decode("ascii")}))
requesting = authorized_request(
api_auth_token,
agent,
@@ -1234,7 +1235,7 @@ class VoucherTests(TestCase):
agent,
b"PUT",
b"http://127.0.0.1/voucher",
- data=BytesIO(dumps({u"voucher": voucher})),
+ data=BytesIO(dumps({u"voucher": voucher.decode("ascii")})),
)
self.assertThat(
putting,
@@ -1361,7 +1362,7 @@ class VoucherTests(TestCase):
note("{} vouchers".format(len(vouchers)))
for voucher in vouchers:
- data = BytesIO(dumps({u"voucher": voucher}))
+ data = BytesIO(dumps({u"voucher": voucher.decode("ascii")}))
putting = authorized_request(
api_auth_token,
agent,
diff --git a/src/_zkapauthorizer/tests/test_controller.py b/src/_zkapauthorizer/tests/test_controller.py
index f84bf05..fce655c 100644
--- a/src/_zkapauthorizer/tests/test_controller.py
+++ b/src/_zkapauthorizer/tests/test_controller.py
@@ -1016,14 +1016,14 @@ def ristretto_verify(signing_key, message, marshaled_passes):
"""
def decode(marshaled_pass):
- t, s = marshaled_pass.split(u" ")
+ t, s = marshaled_pass.split(b" ")
return (
- TokenPreimage.decode_base64(t.encode("ascii")),
- VerificationSignature.decode_base64(s.encode("ascii")),
+ TokenPreimage.decode_base64(t),
+ VerificationSignature.decode_base64(s),
)
servers_passes = list(
- decode(marshaled_pass.pass_text) for marshaled_pass in marshaled_passes
+ decode(marshaled_pass.pass_bytes) for marshaled_pass in marshaled_passes
)
servers_unblinded_tokens = list(
signing_key.rederive_unblinded_token(token_preimage)
diff --git a/src/_zkapauthorizer/tests/test_storage_protocol.py b/src/_zkapauthorizer/tests/test_storage_protocol.py
index 82b62ac..bc11261 100644
--- a/src/_zkapauthorizer/tests/test_storage_protocol.py
+++ b/src/_zkapauthorizer/tests/test_storage_protocol.py
@@ -256,7 +256,7 @@ class ShareTests(TestCase):
# it.
self.local_remote_server.callRemote(
"allocate_buckets",
- list(pass_.pass_text.encode("ascii") for pass_ in all_passes),
+ list(pass_.pass_bytes for pass_ in all_passes),
storage_index,
renew_secret,
cancel_secret,
diff --git a/src/_zkapauthorizer/tests/test_storage_server.py b/src/_zkapauthorizer/tests/test_storage_server.py
index ad570cb..3f5d55a 100644
--- a/src/_zkapauthorizer/tests/test_storage_server.py
+++ b/src/_zkapauthorizer/tests/test_storage_server.py
@@ -80,9 +80,7 @@ class ValidationResultTests(TestCase):
message,
list(RandomToken.create() for i in range(valid_count)),
)
- all_passes = valid_passes + list(
- pass_.pass_text.encode("ascii") for pass_ in invalid_passes
- )
+ all_passes = valid_passes + list(pass_.pass_bytes for pass_ in invalid_passes)
shuffle(all_passes)
self.assertThat(
diff --git a/src/_zkapauthorizer/validators.py b/src/_zkapauthorizer/validators.py
index 0b7284a..5466dd8 100644
--- a/src/_zkapauthorizer/validators.py
+++ b/src/_zkapauthorizer/validators.py
@@ -20,12 +20,17 @@ from base64 import b64decode
def is_base64_encoded(b64decode=b64decode):
+ """
+ Return a a attrs validator that verifies that the attributes is a base64
+ encoded byte string.
+ """
+
def validate_is_base64_encoded(inst, attr, value):
try:
- b64decode(value.encode("ascii"))
+ b64decode(value)
except TypeError:
raise TypeError(
- "{name!r} must be base64 encoded unicode, (got {value!r})".format(
+ "{name!r} must be base64 encoded bytes, (got {value!r})".format(
name=attr.name,
value=value,
),
--
GitLab