Merge pull request #194 from ganwell/t/msgpack-err_e5bf

fix(token): handle errors from msgback by creating a invalid token

manabi/token.py

@@ -13,6 +13,14 @@
 from .util import cattrib, from_string
 
 
+class DecodingError(Exception):
+ pass
+
+
+class EncodingError(Exception):
+ pass
+
+
 def now() -> int:
  return calendar.timegm(datetime.utcnow().timetuple())
 
@@ -104,7 +112,8 @@ def from_ciphertext(cls, key: Key, ciphertext: str) -> "Token":
  return cls(key, None, None)
  try:
  token_path, token_payload = _decode(branca, ciphertext)
- except RuntimeError:
+ except DecodingError:
+ # Handle decoding errors by creating a invalid token
  return cls(key, None, timestamp)
  return cls(key, token_path, token_payload, timestamp, ciphertext)
 
@@ -135,8 +144,18 @@ def _encode(
  now: Optional[int] = None,
 ) -> str:
  f = Branca(key)
- p = umsgpack.packb((path.encode("UTF-8"), payload))
- ciphertext = f.encode(p, now)
+ try:
+ path_bytes = path.encode("UTF-8")
+ except Exception as e:
+ raise EncodingError("Could not UTF-8 encode the path") from e
+ try:
+ p = umsgpack.packb((path_bytes, payload))
+ except Exception as e:
+ raise EncodingError("Could not msg-pack the payload") from e
+ try:
+ ciphertext = f.encode(p, now)
+ except Exception as e:
+ raise EncodingError("Could not encode the branca token") from e
  return ciphertext
 
 
@@ -149,6 +168,16 @@ def _decode(
  f = key
  else:
  f = Branca(key)
- tpb, token_payload = umsgpack.unpackb(f.decode(ciphertext, ttl))
- token_path = tpb.decode("UTF-8")
+ try:
+ token = f.decode(ciphertext, ttl)
+ except Exception as e:
+ raise DecodingError("Could not decode the branca token") from e
+ try:
+ tpb, token_payload = umsgpack.unpackb(token)
+ except Exception as e:
+ raise DecodingError("Could not msg-unpack the payload") from e
+ try:
+ token_path = tpb.decode("UTF-8")
+ except Exception as e:
+ raise DecodingError("Could not UTF-8 decode the path") from e
  return Path(token_path), token_payload

manabi/token_test.py

@@ -9,10 +9,13 @@
 from hypothesis import assume, given, strategies as st
 
 from . import mock
-from .token import TTL, Config, Key, State, Token, _decode, _encode, now
+from .token import TTL, Config, DecodingError, Key, State, Token, _decode, _encode, now
 from .type_alias import OptionalProp
 from .util import from_string
 
+_old_token = "1Ui3IS5xxIedbhSdPFPoGQRnTUtPVTmleMGJe1KyvWsVU704wk68k3YC70txTn5ZEJ4Ms3bh5Esy0OD4mZM0TnumUymWglgp3wq0CHo3W89DyW0"
+_old_key = "bNEZsIjvxDAiLhDA1chvF9zL9OJYPNlCqNPlm7KbhmU"
+
 _key = b"\xef\xc5\x07\xee}\x7f6\x11L\xb0\xc3155x\x11\xce.\x8e\xb96\xba\xce\x8b\x17-\xfc\x96]\xf8%\xd8"
 
 msgpack = st.recursive(
@@ -25,6 +28,12 @@
 )
 
 
+def test_old_token():
+ key = Key(from_string(_old_key))
+ token = Token.from_ciphertext(key, _old_token)
+ assert token.check(0) == State.invalid
+
+
 def test_key_validator(config):
  key = Key.from_dictionary(config)
  assert len(key.data) == 32
@@ -140,7 +149,7 @@ def token_roundtrip(tamper: bool, expire: bool, path: str, payload: OptionalProp
  data = data[0:3] + "f" + data[4:]
 
  if tamper or expire:
- with pytest.raises(RuntimeError):
+ with pytest.raises(DecodingError):
  _decode(key, data, ttl)
  else:
  assert _decode(key, data, ttl) == (Path(path), payload)