Last active
November 17, 2021 11:50
-
-
Save dajiaji/11897ed530ccaec40f04264c98667e82 to your computer and use it in GitHub Desktop.
A simple EUDCC verifier implementation with Python CWT.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import json | |
import os | |
import zlib | |
import requests | |
from base45 import b45decode | |
import cwt | |
from cwt import load_pem_hcert_dsc | |
class Verifier: | |
def __init__(self, base_url: str, trustlist_store_path: str): | |
self._base_url = base_url | |
self._trustlist_store_path = trustlist_store_path | |
self._dscs = [] | |
self._trustlist = [] | |
self._load_trustlist() | |
@classmethod | |
def new(cls, base_url: str, trustlist_store_path: str): | |
return cls(base_url, trustlist_store_path) | |
def refresh_trustlist(self): | |
status = 200 | |
headers = None | |
# Get new DSCs | |
x_resume_token = ( | |
self._trustlist[len(self._trustlist) - 1]["x_resume_token"] | |
if self._trustlist | |
else "" | |
) | |
while status == 200: | |
if x_resume_token: | |
headers = {"X-RESUME-TOKEN": x_resume_token} | |
r = requests.get( | |
self._base_url + "/signercertificateUpdate", headers=headers | |
) | |
status = r.status_code | |
if status == 204: | |
break | |
if status != 200: | |
raise Exception("Received %d from signercertificateUpdate" % status) | |
x_resume_token = r.headers["X-RESUME-TOKEN"] | |
self._trustlist.append( | |
{ | |
"x_kid": r.headers["X-KID"], | |
"x_resume_token": x_resume_token, | |
"dsc": r.text, | |
} | |
) | |
# Filter expired/revoked DSCs | |
r = requests.get(self._base_url + "/signercertificateStatus") | |
if r.status_code != 200: | |
raise Exception("Received %d from signercertificateStatus" % r.status_code) | |
active_kids = r.json() | |
self._dscs = [] | |
for v in self._trustlist: | |
if v["x_kid"] not in active_kids: | |
continue | |
dsc = f"-----BEGIN CERTIFICATE-----\n{v['dsc']}\n-----END CERTIFICATE-----" | |
self._dscs.append(load_pem_hcert_dsc(dsc)) | |
# Update trustlist store. | |
with open(self._trustlist_store_path, "w") as f: | |
json.dump( | |
[v for v in self._trustlist if v["x_kid"] in active_kids], f, indent=4 | |
) | |
return | |
def verify_and_decode(self, eudcc: bytes) -> bytes: | |
if eudcc.startswith(b"HC1:"): | |
# Decode Base45 data. | |
eudcc = b45decode(eudcc[4:]) | |
# Decompress with zlib. | |
eudcc = zlib.decompress(eudcc) | |
# Verify and decode CWT. | |
return cwt.decode(eudcc, keys=self._dscs) | |
def _load_trustlist(self): | |
try: | |
with open(self._trustlist_store_path) as f: | |
self._trustlist = json.load(f) | |
except Exception as err: | |
if type(err) != FileNotFoundError: | |
raise err | |
self._trustlist = [] | |
return | |
# An endpoint of Digital Green Certificate Verifier Service compliant with: | |
# https://eu-digital-green-certificates.github.io/dgca-verifier-service/ | |
BASE_URL = os.environ["CWT_SAMPLES_EUDCC_BASE_URL"] | |
# e.g., "./dscs.json" | |
TRUSTLIST_STORE_PATH = os.environ["CWT_SAMPLES_EUDCC_TRUSTLIST_STORE_PATH"] | |
# quoted from https://github.com/eu-digital-green-certificates/dgc-testdata/blob/main/AT/2DCode/raw/1.json | |
BASE45_FORMATTED_EUDCC = b"HC1:NCFOXN%TS3DH3ZSUZK+.V0ETD%65NL-AH-R6IOOK.IR9B+9G4G50PHZF0AT4V22F/8X*G3M9JUPY0BX/KR96R/S09T./0LWTKD33236J3TA3M*4VV2 73-E3GG396B-43O058YIB73A*G3W19UEBY5:PI0EGSP4*2DN43U*0CEBQ/GXQFY73CIBC:G 7376BXBJBAJ UNFMJCRN0H3PQN*E33H3OA70M3FMJIJN523.K5QZ4A+2XEN QT QTHC31M3+E32R44$28A9H0D3ZCL4JMYAZ+S-A5$XKX6T2YC 35H/ITX8GL2-LH/CJTK96L6SR9MU9RFGJA6Q3QR$P2OIC0JVLA8J3ET3:H3A+2+33U SAAUOT3TPTO4UBZIC0JKQTL*QDKBO.AI9BVYTOCFOPS4IJCOT0$89NT2V457U8+9W2KQ-7LF9-DF07U$B97JJ1D7WKP/HLIJL8JF8JFHJP7NVDEBU1J*Z222E.GJ457661CFFTWM-8P2IUE7K*SSW613:9/:TT5IYQBTBU16R4I1A/9VRPJ-TS.7ZEM7MSVOCD4RG2L-TQJROXL2J:52J7F0Q10SMAP3CG3KHF0DWIH" | |
# RAW_EUDCC = bytes.fromhex( | |
# "d2844da20448d919375fc1e7b6b20126a0590133a4041a61817ca0061a60942ea001624154390103a101a4617681aa62646e01626d616d4f52472d3130303033303231356276706a313131393334393030376264746a323032312d30322d313862636f624154626369783155524e3a555643493a30313a41543a31303830373834334639344145453045453530393346424332353442443831332342626d706c45552f312f32302f31353238626973781b4d696e6973747279206f66204865616c74682c20417573747269616273640262746769383430353339303036636e616da463666e74754d5553544552465241553c474f455353494e47455262666e754d7573746572667261752d47c3b6c39f696e67657263676e74684741425249454c4562676e684761627269656c656376657265312e302e3063646f626a313939382d30322d323658405812fce67cb84c3911d78e3f61f890d0c80eb9675806aebed66aa2d0d0c91d1fc98d7bcb80bf00e181806a9502e11b071325901bd0d2c1b6438747b8cc50f521" | |
# ) | |
if __name__ == "__main__": | |
v = Verifier.new(BASE_URL, TRUSTLIST_STORE_PATH) | |
v.refresh_trustlist() | |
try: | |
res = v.verify_and_decode(BASE45_FORMATTED_EUDCC) # or RAW_EUDCC | |
except Exception as err: | |
print("Verification failed: %s" % err) | |
exit(1) | |
hcert = res[-260] | |
print(hcert) | |
exit(0) |
Is there an official (from the EU?) API available for the DGCA verifier service? Or a list of endpoints? What endpoint was used for this script?
Actually, I am searching for this as well. So far I believe that this is not intended (for whatever reason).
Take a look at EU Digital COVID Certificate: Overview, there it is said:
A note on using the apps and released APK files found in this GitHub organization: The apps are reference implementations that cannot be used in production environments as-is, but rather need to be configured by EU member states to access their national backends. The released APK files are configured to work with the test environments and will not report correct results on "live" DCCs.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Is there an official (from the EU?) API available for the DGCA verifier service? Or a list of endpoints? What endpoint was used for this script?