Add flows for manipulating TUF metadata

Dockerfile

@@ -58,6 +58,7 @@ COPY start.sh /usr/local/bin
 COPY start.py /pipenv
 COPY api /pipenv/api
 COPY flows /pipenv/flows
+COPY fixtures/rsa /unsafe_keys
 
 ENTRYPOINT ["/usr/local/bin/start.sh"]
 EXPOSE 2222 5555

README.md

@@ -4,11 +4,17 @@ This project enables manipulation of [TUF flows](https://theupdateframework.gith
 
 ## Usage
 
-### Starting a transparent proxy
+Run `make` to see the available Makefile commands.
 
-Run `make` to see the available Makefile commands. To boot a new client QEMU image run `make start` with the `IMAGE_DIR` environment variable set to the directory containing the QEMU image.
+### Available proxy flows
 
-By default, `IMAGE_DIR` should contain a QEMU image named `core-image-minimal-qemux86-64.otaimg` and a BIOS file named `u-boot-qemux86-64.rom`, though these can be overridden with `IMAGE_FILE` and `BIOS_FILE` respectively.
+The `flows/` directory contains the individual `mitmproxy` flows that will manipulate the TUF metadata traffic. An HTTP API is started inside Docker to switch between the flows in this directory.
+
+### Starting the transparent proxy
+
+To boot a new client QEMU image run `make start` with the `IMAGE_DIR` environment variable set.
+
+By default, `IMAGE_DIR` should contain a QEMU image named `core-image-minimal-qemux86-64.otaimg` and a BIOS file named `u-boot-qemux86-64.rom`, though these can be overridden by setting `IMAGE_FILE` and `BIOS_FILE` respectively.
 
 ### Controlling the proxy
 

api/datatypes/metadata.py

@@ -55,21 +55,28 @@ def from_readable(cls, data: Readable) -> 'Metadata':
         return cls(json.loads(data))
 
     def to_json(self) -> str:
-        """Output this instance as JSON."""
+        """Return the TUF metadata object as JSON."""
         return str(canonical(self._encode()))
 
+    def canonical_signed(self) -> str:
+        """Return the TUF metadata signed section as JSON."""
+        return str(canonical(self._encode_signed()))
+
     def _encode(self) -> Encoded:
-        out: Dict[str, Any] = {
+        """Encode the signatures and the signed section.."""
+        return {
             "signatures": self.signatures._encode(),
-            "signed": {
-                "_type": self.role,
-                "expires": self.expires,
-                "version": self.version,
-            }
+            "signed": self._encode_signed()
         }
 
+    def _encode_signed(self) -> Encoded:
+        """Encode the signed section."""
+        out: Dict[str, Any] = {
+            "_type": self.role,
+            "expires": self.expires,
+            "version": self.version,
+        }
         if self.targets:
-            out["signed"]["targets"] = self.targets._encode()
-
-        out["signed"].update(self.extra)
+            out["targets"] = self.targets._encode()
+        out.update(self.extra)
         return out

api/datatypes/signature.py

@@ -2,6 +2,7 @@
 
 from base64 import b64decode, b64encode
 from binascii import hexlify
+from copy import deepcopy
 from cytoolz import concat, groupby, remove
 from random import choice
 from rsa import PublicKey
@@ -126,5 +127,19 @@ def replace_random(self, replace_with: Signature) -> 'Signatures':
         """Return a new object with a randomly selected key replaced."""
         return self.replace_key(self.random().keyid, replace_with)
 
+    def duplicate_key(self, key: KeyId) -> 'Signatures':
+        """Return a new object with the matching key duplicated."""
+        matches: Dict[bool, List[Signature]] = groupby(lambda sig: sig.keyid == key, self.sigs)
+        try:
+            sig = matches[True][0]
+            copy = deepcopy(sig)
+            return Signatures(list(concat([matches.get(False, []), [sig, copy]])))
+        except KeyError:
+            return Signatures(self.sigs)
+
+    def duplicate_random(self) -> 'Signatures':
+        """Return a new object with a randomly selected key duplicated."""
+        return self.duplicate_key(self.random().keyid)
+
     def _encode(self) -> Encoded:
         return [sig._encode() for sig in self.sigs]  # type: ignore

api/datatypes/signature_test.py

@@ -36,3 +36,10 @@ def test_sigs_replace_key() -> None:
     assert sigs.find(rand_key) is None
     assert sigs.find(new_sig.keyid) is not None
     assert len(sigs.sigs) == 3
+
+def test_sigs_duplicate_key() -> None:
+    sigs = Signatures([sig() for sig in repeat(random_sig, 3)])
+    rand_key = sigs.random().keyid
+    sigs = sigs.duplicate_key(rand_key)
+    assert sigs.find(rand_key) is not None
+    assert len(sigs.sigs) == 4

api/utils.py

@@ -1,6 +1,7 @@
 import json
 
 from json import JSONEncoder
+from mitmproxy.http import HTTPFlow
 from typing import Any, Dict, List, Union
 from typing_extensions import Protocol
 
@@ -31,3 +32,13 @@ def contains(meta: Dict[str, Any], *fields: str) -> None:
 def canonical(encoded: Encoded) -> str:
     """Canonicalize the encoded object as a JSON string."""
     return json.dumps(encoded, sort_keys=True, separators=(',', ':'), cls=Encoder)
+
+
+def is_metadata(flow: HTTPFlow) -> bool:
+    """Inspect the flow to identify whether it is TUF metadata."""
+    if flow.response.headers.get('Content-Type') != "application/json":
+        return False
+    elif flow.response.text == "{}":
+        return False
+    else:
+        return True

flows/delete_signature.py

@@ -0,0 +1,25 @@
+from mitmproxy import ctx
+from mitmproxy.http import HTTPFlow
+
+from api.datatypes.metadata import Metadata
+from api.utils import is_metadata
+
+
+def response(flow: HTTPFlow) -> None:
+    if is_metadata(flow):
+        ctx.log.info(f"Deleting a signature...")
+    else:
+        ctx.log.debug("skipping non-metadata response...")
+        return
+
+    try:
+        meta = Metadata.from_flow(flow)
+        del_sig = meta.signatures.random()
+        ctx.log.debug(f"deleting sig with keyid: {del_sig.keyid}")
+        meta.signatures = meta.signatures.remove_key(del_sig.keyid)
+
+        flow.response.headers["x-mitm-flow"] = "delete_signature"
+        flow.response.content = meta.to_json().encode("UTF-8")
+    except Exception as e:
+        ctx.log.error(f"Processing error: {e}")
+        ctx.log.debug(e.__traceback__)

flows/duplicate_signature.py

@@ -0,0 +1,25 @@
+from mitmproxy import ctx
+from mitmproxy.http import HTTPFlow
+
+from api.datatypes.metadata import Metadata
+from api.utils import is_metadata
+
+
+def response(flow: HTTPFlow) -> None:
+    if is_metadata(flow):
+        ctx.log.info(f"Duplicating a signature...")
+    else:
+        ctx.log.debug("skipping non-metadata response...")
+        return
+
+    try:
+        meta = Metadata.from_flow(flow)
+        dup_sig = meta.signatures.random()
+        ctx.log.debug(f"duplicating sig with keyid: {dup_sig.keyid}")
+        meta.signatures = meta.signatures.duplicate_key(dup_sig.keyid)
+
+        flow.response.headers["x-mitm-flow"] = "duplicate_signature"
+        flow.response.content = meta.to_json().encode("UTF-8")
+    except Exception as e:
+        ctx.log.error(f"Processing error: {e}")
+        ctx.log.debug(e.__traceback__)

flows/new_signature.py

@@ -0,0 +1,34 @@
+from mitmproxy import ctx
+from mitmproxy.http import HTTPFlow
+
+from api.datatypes.metadata import Metadata
+from api.datatypes.signing import Rsa
+from api.utils import is_metadata
+
+
+PUB_KEY = "/unsafe_keys/rsa_4096.pub"
+PRIV_KEY = "/unsafe_keys/rsa_4096.key"
+
+def response(flow: HTTPFlow) -> None:
+    if is_metadata(flow):
+        ctx.log.info(f"Replacing a signature with one from another key...")
+    else:
+        ctx.log.debug("skipping non-metadata response...")
+        return
+
+    try:
+        meta = Metadata.from_flow(flow)
+        rsa = Rsa.from_files(PUB_KEY, PRIV_KEY)
+
+        sigs = meta.signatures
+        old_sig = sigs.random()
+        ctx.log.debug(f"deleting sig with keyid: {old_sig.keyid}")
+        new_sig = rsa.sign(meta.canonical_signed().encode("UTF-8"))
+        ctx.log.debug(f"adding sig with keyid: {new_sig.keyid}")
+        meta.signatures = sigs.replace_key(old_sig.keyid, new_sig)
+
+        flow.response.headers["x-mitm-flow"] = "new_signature"
+        flow.response.content = meta.to_json().encode("UTF-8")
+    except Exception as e:
+        ctx.log.error(f"Processing error: {e}")
+        ctx.log.debug(e.__traceback__)

flows/passthrough.py

@@ -0,0 +1,11 @@
+from mitmproxy import ctx
+from mitmproxy.http import HTTPFlow
+
+from api.utils import is_metadata
+
+
+def response(flow: HTTPFlow) -> None:
+    if is_metadata(flow):
+        ctx.log.debug("skipping metadata response...")
+    else:
+        ctx.log.debug("skipping non-metadata response...")

flows/randomize_keyid.py

@@ -0,0 +1,26 @@
+from mitmproxy import ctx
+from mitmproxy.http import HTTPFlow
+
+from api.datatypes.metadata import Metadata
+from api.utils import is_metadata
+
+
+def response(flow: HTTPFlow) -> None:
+    if is_metadata(flow):
+        ctx.log.info(f"Randomizing a key-id...")
+    else:
+        ctx.log.debug("skipping non-metadata response...")
+        return
+
+    try:
+        meta = Metadata.from_flow(flow)
+        old_sig = meta.signatures.random()
+        new_sig = old_sig.randomize_key()
+        ctx.log.debug(f"changing key-id from {old_sig.keyid} to {new_sig.keyid}...")
+        meta.signatures = meta.signatures.replace_key(old_sig.keyid, new_sig)
+
+        flow.response.headers["x-mitm-flow"] = "randomize_keyid"
+        flow.response.content = meta.to_json().encode("UTF-8")
+    except Exception as e:
+        ctx.log.error(f"Processing error: {e}")
+        ctx.log.debug(e.__traceback__)

flows/randomize_signature.py

@@ -0,0 +1,26 @@
+from mitmproxy import ctx
+from mitmproxy.http import HTTPFlow
+
+from api.datatypes.metadata import Metadata
+from api.utils import is_metadata
+
+
+def response(flow: HTTPFlow) -> None:
+    if is_metadata(flow):
+        ctx.log.info(f"Randomizing a signature...")
+    else:
+        ctx.log.debug("skipping non-metadata response...")
+        return
+
+    try:
+        meta = Metadata.from_flow(flow)
+        old_sig = meta.signatures.random()
+        new_sig = old_sig.randomize_sig()
+        ctx.log.debug(f"replacing keyid {old_sig.keyid} with {new_sig}.keyid")
+        meta.signatures = meta.signatures.replace_key(old_sig.keyid, new_sig)
+
+        flow.response.headers["x-mitm-flow"] = "randomize_signature"
+        flow.response.content = meta.to_json().encode("UTF-8")
+    except Exception as e:
+        ctx.log.error(f"Processing error: {e}")
+        ctx.log.debug(e.__traceback__)

flows/randomize_version.py

@@ -0,0 +1,28 @@
+import sys
+
+from mitmproxy import ctx
+from mitmproxy.http import HTTPFlow
+from random import randrange
+
+from api.datatypes.metadata import Metadata
+from api.utils import is_metadata
+
+
+def response(flow: HTTPFlow) -> None:
+    if is_metadata(flow):
+        ctx.log.info(f"Randomize the signed version...")
+    else:
+        ctx.log.debug("skipping non-metadata response...")
+        return
+
+    try:
+        meta = Metadata.from_flow(flow)
+        new_version = randrange(sys.maxsize)
+        ctx.log.debug(f"replacing metadata version {meta.version} with {new_version}")
+        meta.version = new_version
+
+        flow.response.headers["x-mitm-flow"] = "randomize_version"
+        flow.response.content = meta.to_json().encode("UTF-8")
+    except Exception as e:
+        ctx.log.error(f"Processing error: {e}")
+        ctx.log.debug(e.__traceback__)

start.sh

@@ -82,7 +82,7 @@ info "Starting the API..." && {
     --http.host="${HTTP_HOST:-0.0.0.0}" \
     --http.port="${HTTP_PORT:-5555}" \
     --flow.root="${FLOW_ROOT:-/pipenv/flows}" \
-    --flow.initial="${FLOW_INITIAL:-random_sig}" \
+    --flow.initial="${FLOW_INITIAL:-passthrough}" \
     --mitm.cadir=/certs \
     --mitm.upstream_trusted_ca="/certs/${ROOT_CERT}" \
     --mitm.client_certs="/certs/bundle.pem"