Skip to content

Instantly share code, notes, and snippets.

@DavidBuchanan314

DavidBuchanan314/google_c2pa_ra_sign_poc.py

Last active September 15, 2026 14:14
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save DavidBuchanan314/fa0ffdaaaa31594e6a511118c1cea1e0 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save DavidBuchanan314/fa0ffdaaaa31594e6a511118c1cea1e0 to your computer and use it in GitHub Desktop.
Download ZIP
See https://www.da.vidbuchanan.co.uk/blog/android-c2pa.html - protobuf reversing and code herein generated by Opus 5 (via jadx-decompiled Pixel Camera sources)
Raw
google_c2pa_ra_sign_poc.py
import base64
import json
import sys
import threading
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import c2pa # python3 -m pip install c2pa-python
import keystork # https://github.com/DavidBuchanan314/keystork
import requests
from asn1crypto import csr as asn1_csr, keys as asn1_keys, x509 as asn1_x509
from cryptography import x509
from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
import google_ra_pb2 as pb
PACKAGE = "com.google.android.GoogleCamera"
BASE = "https://c2paregistration.pa.googleapis.com/v0alpha"
HEADERS = {
"Content-Type": "application/x-protobuf",
"Accept": "application/x-protobuf",
"X-Goog-Api-Key": "AIzaSyB7rueliYAu9A89LPMDRi7oUPQ6F9VDJkE",
}
TSA = "https://coretimestamping.pa.googleapis.com/v0alpha/getTimestamp"
TSA_HEADERS = {
"Content-Type": "application/x-protobuf",
"x-goog-api-key": "AIzaSyCoP3r8qAZuhuCCZT7iMz_zzmQhdTUf814",
}
SECURITY_LEVEL = keystork.SecurityLevel.STRONGBOX
DIGITAL_CAPTURE = "http://cv.iptc.org/newscodes/digitalsourcetype/digitalCapture"
source, dest = sys.argv[1], sys.argv[2]
MANIFEST = {
"claim_generator": "c2pa poc",
"claim_generator_info": [{"name": "c2pa poc"}],
"title": dest,
"assertions": [
{
"label": "c2pa.actions",
"data": {
"actions": [
{
"action": "c2pa.created",
"digitalSourceType": DIGITAL_CAPTURE,
"softwareAgent": "c2pa poc",
}
]
},
}
],
}
def post(method, message):
url = f"{BASE}/{method}"
print(f"[*] POST {url}")
print(">", message)
response = requests.post(
url, data=message.SerializeToString(), headers=HEADERS
)
response.raise_for_status()
return response.content
def rdn(kind, value):
return asn1_x509.RelativeDistinguishedName([
asn1_x509.NameTypeAndValue({
"type": kind,
"value": asn1_x509.DirectoryString({"utf8_string": value}),
})
])
def pem(der):
body = base64.b64encode(der).decode()
lines = "\n".join(body[i:i + 64] for i in range(0, len(body), 64))
return f"-----BEGIN CERTIFICATE-----\n{lines}\n-----END CERTIFICATE-----\n"
challenge = pb.GetChallengeResponse.FromString(
post("getChallenge", pb.GetChallengeRequest(package_name=PACKAGE))
)
print("<", challenge)
nonce = challenge.challenge.nonce
profile = challenge.certificate_profile.profile
rdns = [
rdn("common_name", profile.common_name),
rdn("organization_name", profile.organization),
]
if profile.organizational_unit:
rdns.append(rdn("organizational_unit_name", profile.organizational_unit))
alias = str(uuid.uuid4())
device = keystork.Device()
with device.connect() as conn:
def prop(name):
vendor = conn.check_output(["/system/bin/getprop", f"ro.product.vendor.{name}"]).decode().strip()
return vendor or conn.check_output(["/system/bin/getprop", f"ro.product.{name}"]).decode().strip()
device_ids = {
keystork.Tag.ATTESTATION_ID_BRAND: prop("brand").encode(),
keystork.Tag.ATTESTATION_ID_DEVICE: prop("device").encode(),
keystork.Tag.ATTESTATION_ID_PRODUCT: prop("name").encode(),
keystork.Tag.ATTESTATION_ID_MANUFACTURER: prop("manufacturer").encode(),
keystork.Tag.ATTESTATION_ID_MODEL: prop("model").encode(),
}
uid = keystork.resolve_uid(conn, PACKAGE)
with conn.open_keystore_session(uid) as session:
metadata = session.generate_key_pair(
alias,
purposes=(keystork.KeyPurpose.SIGN,),
key_size=256,
ec_curve=keystork.EcCurve.P_256,
attestation_challenge=nonce,
device_ids=device_ids,
security_level=SECURITY_LEVEL,
)
public_key = x509.load_der_x509_certificate(metadata.certificates[0]).public_key()
info = asn1_csr.CertificationRequestInfo({
"version": "v1",
"subject": asn1_x509.Name(name="", value=asn1_x509.RDNSequence(rdns)),
"subject_pk_info": asn1_keys.PublicKeyInfo.load(
public_key.public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)
),
"attributes": [],
})
signature = session.sign(alias, info.dump(), security_level=SECURITY_LEVEL)
csr = asn1_csr.CertificationRequest({
"certification_request_info": info,
"signature_algorithm": {"algorithm": "sha256_ecdsa"},
"signature": signature,
})
request = pb.EnrollAndroidRequest()
enrollment = request.enrollments.add()
enrollment.csr = csr.dump()
enrollment.attestation_chain.extend(metadata.certificates)
response = pb.EnrollAndroidResponse.FromString(post("enrollAndroid", request))
print("<", response)
issued = response.issued[0]
chain = "".join(pem(der) for der in [issued.leaf[0], *issued.ca_chain])
print("[+] provisioned cert chain:")
print(chain)
def sign_claim(data):
with device.connect() as conn:
with conn.open_keystore_session(uid) as session:
der = session.sign(alias, data, security_level=SECURITY_LEVEL)
r, s = decode_dss_signature(der)
return r.to_bytes(32, "big") + s.to_bytes(32, "big")
class TimestampProxy(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def do_POST(self):
query = self.rfile.read(int(self.headers["Content-Length"]))
upstream = requests.post(
TSA,
data=pb.GetTimestampRequest(timestamp_request=query).SerializeToString(),
headers=TSA_HEADERS,
)
upstream.raise_for_status()
reply = pb.GetTimestampResponse.FromString(upstream.content).timestamp_response
self.send_response(200)
self.send_header("Content-Type", "application/timestamp-reply")
self.send_header("Content-Length", str(len(reply)))
self.end_headers()
self.wfile.write(reply)
def log_message(self, *args):
pass
proxy = ThreadingHTTPServer(("127.0.0.1", 0), TimestampProxy)
threading.Thread(target=proxy.serve_forever, daemon=True).start()
tsa_url = f"http://127.0.0.1:{proxy.server_port}/"
print("[*] timestamp proxy:", tsa_url, "->", TSA)
context = c2pa.Context.from_dict(
{"builder": {"created_assertion_labels": ["c2pa.actions", "c2pa.actions.v2"]}}
)
signer = c2pa.Signer.from_callback(sign_claim, c2pa.C2paSigningAlg.ES256, chain, tsa_url)
c2pa.Builder(json.dumps(MANIFEST), context=context).sign_file(source, dest, signer)
proxy.shutdown()
print("[+] signed:", source, "->", dest)
Raw
google_ra.proto
syntax = "proto3";
package c2papocs.googlera;
import "google/protobuf/timestamp.proto";
message GetChallengeRequest {
string package_name = 1;
string conformance_record = 2;
}
message GetChallengeResponse {
Challenge challenge = 1;
CertificateProfileWrapper certificate_profile = 2;
}
message Challenge {
bytes nonce = 1;
google.protobuf.Timestamp expiration = 2;
}
message CertificateProfileWrapper {
CertificateProfile profile = 1;
}
message CertificateProfile {
string common_name = 1;
string organization = 2;
string organizational_unit = 3;
}
message EnrollAndroidRequest {
repeated AndroidEnrollment enrollments = 4;
}
message AndroidEnrollment {
bytes csr = 1;
repeated bytes attestation_chain = 2;
}
message EnrollAndroidResponse {
repeated IssuedCertificates issued = 2;
}
message IssuedCertificates {
repeated bytes leaf = 1;
repeated bytes ca_chain = 2;
}
// protobuf-wrapped RFC 3161
message GetTimestampRequest {
bytes timestamp_request = 1;
}
message GetTimestampResponse {
bytes timestamp_response = 1;
}
message Status {
int32 code = 1;
string message = 2;
}
Raw
google_ra_pb2.py
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler. DO NOT EDIT!
# source: google_ra.proto
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import message as _message
from google.protobuf import reflection as _reflection
from google.protobuf import symbol_database as _symbol_database
# @@protoc_insertion_point(imports)
_sym_db = _symbol_database.Default()
from google.protobuf import timestamp_pb2 as google_dot_protobuf_dot_timestamp__pb2
DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1f\x63\x32pa_pocs/proto/google_ra.proto\x12\x11\x63\x32papocs.googlera\x1a\x1fgoogle/protobuf/timestamp.proto\"G\n\x13GetChallengeRequest\x12\x14\n\x0cpackage_name\x18\x01 \x01(\t\x12\x1a\n\x12\x63onformance_record\x18\x02 \x01(\t\"\x92\x01\n\x14GetChallengeResponse\x12/\n\tchallenge\x18\x01 \x01(\x0b\x32\x1c.c2papocs.googlera.Challenge\x12I\n\x13\x63\x65rtificate_profile\x18\x02 \x01(\x0b\x32,.c2papocs.googlera.CertificateProfileWrapper\"J\n\tChallenge\x12\r\n\x05nonce\x18\x01 \x01(\x0c\x12.\n\nexpiration\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\"S\n\x19\x43\x65rtificateProfileWrapper\x12\x36\n\x07profile\x18\x01 \x01(\x0b\x32%.c2papocs.googlera.CertificateProfile\"\\\n\x12\x43\x65rtificateProfile\x12\x13\n\x0b\x63ommon_name\x18\x01 \x01(\t\x12\x14\n\x0corganization\x18\x02 \x01(\t\x12\x1b\n\x13organizational_unit\x18\x03 \x01(\t\"Q\n\x14\x45nrollAndroidRequest\x12\x39\n\x0b\x65nrollments\x18\x04 \x03(\x0b\x32$.c2papocs.googlera.AndroidEnrollment\";\n\x11\x41ndroidEnrollment\x12\x0b\n\x03\x63sr\x18\x01 \x01(\x0c\x12\x19\n\x11\x61ttestation_chain\x18\x02 \x03(\x0c\"N\n\x15\x45nrollAndroidResponse\x12\x35\n\x06issued\x18\x02 \x03(\x0b\x32%.c2papocs.googlera.IssuedCertificates\"4\n\x12IssuedCertificates\x12\x0c\n\x04leaf\x18\x01 \x03(\x0c\x12\x10\n\x08\x63\x61_chain\x18\x02 \x03(\x0c\"0\n\x13GetTimestampRequest\x12\x19\n\x11timestamp_request\x18\x01 \x01(\x0c\"2\n\x14GetTimestampResponse\x12\x1a\n\x12timestamp_response\x18\x01 \x01(\x0c\x62\x06proto3')
_GETCHALLENGEREQUEST = DESCRIPTOR.message_types_by_name['GetChallengeRequest']
_GETCHALLENGERESPONSE = DESCRIPTOR.message_types_by_name['GetChallengeResponse']
_CHALLENGE = DESCRIPTOR.message_types_by_name['Challenge']
_CERTIFICATEPROFILEWRAPPER = DESCRIPTOR.message_types_by_name['CertificateProfileWrapper']
_CERTIFICATEPROFILE = DESCRIPTOR.message_types_by_name['CertificateProfile']
_ENROLLANDROIDREQUEST = DESCRIPTOR.message_types_by_name['EnrollAndroidRequest']
_ANDROIDENROLLMENT = DESCRIPTOR.message_types_by_name['AndroidEnrollment']
_ENROLLANDROIDRESPONSE = DESCRIPTOR.message_types_by_name['EnrollAndroidResponse']
_ISSUEDCERTIFICATES = DESCRIPTOR.message_types_by_name['IssuedCertificates']
_GETTIMESTAMPREQUEST = DESCRIPTOR.message_types_by_name['GetTimestampRequest']
_GETTIMESTAMPRESPONSE = DESCRIPTOR.message_types_by_name['GetTimestampResponse']
GetChallengeRequest = _reflection.GeneratedProtocolMessageType('GetChallengeRequest', (_message.Message,), {
'DESCRIPTOR' : _GETCHALLENGEREQUEST,
'__module__' : 'c2pa_pocs.proto.google_ra_pb2'
# @@protoc_insertion_point(class_scope:c2papocs.googlera.GetChallengeRequest)
})
_sym_db.RegisterMessage(GetChallengeRequest)
GetChallengeResponse = _reflection.GeneratedProtocolMessageType('GetChallengeResponse', (_message.Message,), {
'DESCRIPTOR' : _GETCHALLENGERESPONSE,
'__module__' : 'c2pa_pocs.proto.google_ra_pb2'
# @@protoc_insertion_point(class_scope:c2papocs.googlera.GetChallengeResponse)
})
_sym_db.RegisterMessage(GetChallengeResponse)
Challenge = _reflection.GeneratedProtocolMessageType('Challenge', (_message.Message,), {
'DESCRIPTOR' : _CHALLENGE,
'__module__' : 'c2pa_pocs.proto.google_ra_pb2'
# @@protoc_insertion_point(class_scope:c2papocs.googlera.Challenge)
})
_sym_db.RegisterMessage(Challenge)
CertificateProfileWrapper = _reflection.GeneratedProtocolMessageType('CertificateProfileWrapper', (_message.Message,), {
'DESCRIPTOR' : _CERTIFICATEPROFILEWRAPPER,
'__module__' : 'c2pa_pocs.proto.google_ra_pb2'
# @@protoc_insertion_point(class_scope:c2papocs.googlera.CertificateProfileWrapper)
})
_sym_db.RegisterMessage(CertificateProfileWrapper)
CertificateProfile = _reflection.GeneratedProtocolMessageType('CertificateProfile', (_message.Message,), {
'DESCRIPTOR' : _CERTIFICATEPROFILE,
'__module__' : 'c2pa_pocs.proto.google_ra_pb2'
# @@protoc_insertion_point(class_scope:c2papocs.googlera.CertificateProfile)
})
_sym_db.RegisterMessage(CertificateProfile)
EnrollAndroidRequest = _reflection.GeneratedProtocolMessageType('EnrollAndroidRequest', (_message.Message,), {
'DESCRIPTOR' : _ENROLLANDROIDREQUEST,
'__module__' : 'c2pa_pocs.proto.google_ra_pb2'
# @@protoc_insertion_point(class_scope:c2papocs.googlera.EnrollAndroidRequest)
})
_sym_db.RegisterMessage(EnrollAndroidRequest)
AndroidEnrollment = _reflection.GeneratedProtocolMessageType('AndroidEnrollment', (_message.Message,), {
'DESCRIPTOR' : _ANDROIDENROLLMENT,
'__module__' : 'c2pa_pocs.proto.google_ra_pb2'
# @@protoc_insertion_point(class_scope:c2papocs.googlera.AndroidEnrollment)
})
_sym_db.RegisterMessage(AndroidEnrollment)
EnrollAndroidResponse = _reflection.GeneratedProtocolMessageType('EnrollAndroidResponse', (_message.Message,), {
'DESCRIPTOR' : _ENROLLANDROIDRESPONSE,
'__module__' : 'c2pa_pocs.proto.google_ra_pb2'
# @@protoc_insertion_point(class_scope:c2papocs.googlera.EnrollAndroidResponse)
})
_sym_db.RegisterMessage(EnrollAndroidResponse)
IssuedCertificates = _reflection.GeneratedProtocolMessageType('IssuedCertificates', (_message.Message,), {
'DESCRIPTOR' : _ISSUEDCERTIFICATES,
'__module__' : 'c2pa_pocs.proto.google_ra_pb2'
# @@protoc_insertion_point(class_scope:c2papocs.googlera.IssuedCertificates)
})
_sym_db.RegisterMessage(IssuedCertificates)
GetTimestampRequest = _reflection.GeneratedProtocolMessageType('GetTimestampRequest', (_message.Message,), {
'DESCRIPTOR' : _GETTIMESTAMPREQUEST,
'__module__' : 'c2pa_pocs.proto.google_ra_pb2'
# @@protoc_insertion_point(class_scope:c2papocs.googlera.GetTimestampRequest)
})
_sym_db.RegisterMessage(GetTimestampRequest)
GetTimestampResponse = _reflection.GeneratedProtocolMessageType('GetTimestampResponse', (_message.Message,), {
'DESCRIPTOR' : _GETTIMESTAMPRESPONSE,
'__module__' : 'c2pa_pocs.proto.google_ra_pb2'
# @@protoc_insertion_point(class_scope:c2papocs.googlera.GetTimestampResponse)
})
_sym_db.RegisterMessage(GetTimestampResponse)
if _descriptor._USE_C_DESCRIPTORS == False:
DESCRIPTOR._options = None
_GETCHALLENGEREQUEST._serialized_start=87
_GETCHALLENGEREQUEST._serialized_end=158
_GETCHALLENGERESPONSE._serialized_start=161
_GETCHALLENGERESPONSE._serialized_end=307
_CHALLENGE._serialized_start=309
_CHALLENGE._serialized_end=383
_CERTIFICATEPROFILEWRAPPER._serialized_start=385
_CERTIFICATEPROFILEWRAPPER._serialized_end=468
_CERTIFICATEPROFILE._serialized_start=470
_CERTIFICATEPROFILE._serialized_end=562
_ENROLLANDROIDREQUEST._serialized_start=564
_ENROLLANDROIDREQUEST._serialized_end=645
_ANDROIDENROLLMENT._serialized_start=647
_ANDROIDENROLLMENT._serialized_end=706
_ENROLLANDROIDRESPONSE._serialized_start=708
_ENROLLANDROIDRESPONSE._serialized_end=786
_ISSUEDCERTIFICATES._serialized_start=788
_ISSUEDCERTIFICATES._serialized_end=840
_GETTIMESTAMPREQUEST._serialized_start=842
_GETTIMESTAMPREQUEST._serialized_end=890
_GETTIMESTAMPRESPONSE._serialized_start=892
_GETTIMESTAMPRESPONSE._serialized_end=942
# @@protoc_insertion_point(module_scope)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment