Anpassungen nach dem ersten Praxiseinsatz

Möglichkeit der Webcamspiegelung für einfachere QR-Code-Positionierung,
besseres Namensmatching der Teilnehmer,
automatisches löschen der Kontaktlisten

.gitignore

 __pycache__/
+contacts/
+attendees/
 *.pem
 *.crt
 *.json
-*.csv
-*.log
-*.err

CovPass-Check.desktop

+#!/usr/bin/env xdg-open
+[Desktop Entry]
+Version=1.0
+Type=Application
+Terminal=false
+Exec=/opt/covpass-check/start.sh
+Name=CovPass-Check
+Comment=CovPass-Check
+Icon=/opt/covpass-check/icon.png

README.md

 CovPass Check
 =============
 
-Im WS21/22 müssen min. 10% der Teilnehmer von Lehrveranstaltungen bis zu 50 Teilnehmern getestet werden.
+Im WS21/22 müssen min. 10% der Teilnehmer von Lehrveranstaltungen bis zu 50
+Teilnehmern getestet werden.
 
-Wie wäre es, wenn wir einfach am Eingang einen Notebook mit Webcam hinstellen, und Studis selbstständig ihren Code scannen lassen?
+Wie wäre es, wenn wir einfach am Eingang einen Notebook mit Webcam hinstellen,
+und Studis selbstständig ihren Code scannen lassen?
 
 ![Bildschirmausgabe](screenshot.jpg)
 
-Auf dem Bildschirm wird die Webcamausgabe angezeigt, erkannte QR Codes werden farblich markiert:
+Auf dem Bildschirm wird die Webcamausgabe angezeigt, erkannte QR Codes werden
+farblich markiert:
 
  * blau: Noch nicht verarbeitet
  * grün (mit Namen): Erfolgreich verifizierter QR-Code
  * gelb (mit Namen): verifizierter QR-Code eines nicht angemeldeten Teilnehmers (siehe unten)
  * rot: Invalider QR-Code oder ungültiger Ausweis
 
-Die Anzahl der erfolgreichen (eindeutigen) Eingaben wird oben links in der Ecke angezeigt (im Beispiel: 2) -- im Idealfall entspricht es der Anzahl der Personen im Raum.
+Die Anzahl der erfolgreichen (eindeutigen) Eingaben wird oben links in der Ecke
+angezeigt (im Beispiel: 2) -- im Idealfall entspricht es der Anzahl der Personen
+im Raum.
 
 
 **Problem:** (Personal)Ausweise werden nicht kontrolliert.
 
-Somit könnte z.B. von der Oma der Impfausweis im Handy abgespeichert und vorgezeigt werden.
+Somit könnte z.B. von der Oma der Impfausweis im Handy abgespeichert und
+vorgezeigt werden.
 
-*Mögliche Lösung:* Automatischer Abgleich mit den Personendaten aus dem dazugehörigen Waffelkurs.
-Dazu einfach eine Datei mit zeilenweise Nachname und Vorname (getrennt durch ein `;`) der angemeldeter Teilnehmer erstellen -- schon werden Namen, welche sich nicht auf der Liste befinden, in gelber Farbe angezeigt und nicht mit gezählt.
+*Mögliche Lösung:* Automatischer Abgleich mit den Personendaten aus dem
+dazugehörigen Waffelkurs.
+Dazu einfach eine Datei mit zeilenweise Nachname und Vorname (getrennt durch ein
+`;`) der angemeldeter Teilnehmer erstellen -- schon werden Namen, welche sich
+nicht auf der Liste befinden, in gelber Farbe angezeigt und nicht mit gezählt.
 
 Nicht perfekt, aber besser als nichts.
 
@@ -45,17 +54,21 @@ Weitere Informationen via
 
     python3 webcam.py -h
 
-Für akustische Teilnehmerbeschränkung bei einer geringen Webcamauflösung beispielsweise
+Für akustische Teilnehmerbeschränkung bei einer geringen Webcamauflösung
+beispielsweise
 
-    python3 webcam.py -a students.txt -s -r 640x480 -l log.txt
+    python3 webcam.py -a students.csv -s -r 640x480 -l log.txt
 
 Für den Übungsbetrieb einfach die Teilnehmerliste(n) der Veranstaltung(en) im
-oben beschriebenen`.csv` Format in den Ordner legen und schlicht
+oben beschriebenen`.csv` Format in den Unterordner `attendees` legen und schlicht
 
     ./start.sh
 
 ausführen (bei Bedarf die Parameter in der Variable `PARAMS` anpassen).
 
+Das Skript hält eine Kontaktliste für 2 Wochen im Unterordner `contacts` bereit
+(ältere Listen werden automatisch gelöscht).
+
 
 Weiterführende Informationen
 ----------------------------

start.sh

 #!/bin/bash
-PARAMS="-s -f"
+# Parameter: akkustische Mitteilung, vollbild, spiegeln
+PARAMS="-s -f -m -Q qrcodes.txt"
 
+
+# Wechsle in Skriptverzeichnis
+cd "$(dirname "${BASH_SOURCE[0]}")"
+
+# Dateieinstellungen
+TRUSTLIST='trustlist.json'
+TRUSTLIST_MAX_AGE='7 days'
+# Teilnehmerliste
+ATTENDEEDIR='attendees'
+# Kontaktliste
+CONTACTDIR='contacts'
+CONTACTLIST_MAX_AGE='15 days'
+
+
+# Lade Wurzelzertifikate
 export CH_TOKEN='0795dc8b-d8d0-4313-abf2-510b12d50939'
 export FR_TOKEN='eyJhbGciOiJSUzI1NiIsInR5cCIgOiAiSldUIiwia2lkIiA6ICJqUG0zZ1BzUlZaMWRRUmhHOG1HMGhFN3Jlb2ZXTTNINzJCV1RtajdJcFd3In0.eyJleHAiOjE2ODUxODU5MDYsImlhdCI6MTYyMjExMzkwNiwianRpIjoiOTdjODgyM2EtNjlhZS00NzA4LWE4N2UtNzYxM2NhNGU3ODU5IiwiaXNzIjoiaHR0cHM6Ly9hdXRoLm1lc3NlcnZpY2VzLmluZ3JvdXBlLmNvbS9hdXRoL3JlYWxtcy9QSU5HIiwiYXVkIjoiYWNjb3VudCIsInN1YiI6ImVhMWY1NWVlLTUxMGMtNGMxNi05MWQ4LTE1MjI4OGJhZDViYSIsInR5cCI6IkJlYXJlciIsImF6cCI6InRhY3YtY2xpZW50LWxpdGUiLCJzZXNzaW9uX3N0YXRlIjoiNjk5ODExY2YtODFlZS00ZmNkLTk4NDctY2FkMGJmYjZhOTdiIiwiYWNyIjoiMSIsInJlYWxtX2FjY2VzcyI6eyJyb2xlcyI6WyJST0xFX1ZFUklGWV9DT05UUk9MXzJERE9DX0wxIiwiUk9MRV9WRVJJRllfQ09OVFJPTF8yRERPQ19CMiIsIm9mZmxpbmVfYWNjZXNzIiwidW1hX2F1dGhvcml6YXRpb24iXX0sInJlc291cmNlX2FjY2VzcyI6eyJhY2NvdW50Ijp7InJvbGVzIjpbIm1hbmFnZS1hY2NvdW50IiwibWFuYWdlLWFjY291bnQtbGlua3MiLCJ2aWV3LXByb2ZpbGUiXX19LCJzY29wZSI6ImVtYWlsIHByb2ZpbGUgb2ZmbGluZV9hY2Nlc3MiLCJzaXJlbiI6IjAwMDAwMDAwMCIsImVtYWlsX3ZlcmlmaWVkIjpmYWxzZSwicHJlZmVycmVkX3VzZXJuYW1lIjoidGFjdi1tb2JpbGUtbGl0ZSIsImdpdmVuX25hbWUiOiIiLCJmYW1pbHlfbmFtZSI6IiJ9.mpfrIP8ayElTm7yoVayCF11oYrDQEnauk9hbbVBw8idAiE6OsMlWNloZtUbbnwrJZsMX3_NoEyzkiB3HNbxyhPWp7eRZ7qhn8XjZVgg6sVytXqcVZo9R5-Q9JftMKv7JelsY3PsaOo5x-pYOX30ancPRjd78TeenorGopsVN_LLRLQpenfgjjgwx-srZnLa-TFYTcbSvXozfJT7uk5CHyz_MIFLM7pl9Zdt66yTGBkLIyOLFsV5vPeH5SYvgRNDYdxZy4XMo6Gyfz0lAI9Xfcjs20NBoOQMV4JREH4Z-IcJJXeszC9QeA1-tRmxujqIRuyvBal7msLy7Zimd2q7i3Q' 
 
@@ -18,8 +34,7 @@ for ROOT_CERT in *.pem ; do
 	fi
 done
 
-TRUSTLIST='trustlist.json'
-TRUSTLIST_MAX_AGE='10 days'
+# Cache Trustlist
 if [[ ! -f "$TRUSTLIST" ]] ; then
 	echo "Downloading (new) trust list"
 	python3 verify_ehc.py --certs-from $CERTS --save-certs $TRUSTLIST 
@@ -28,12 +43,26 @@ elif [[ $(date --date="$TRUSTLIST_MAX_AGE ago" +%s) -gt $(date -r "$TRUSTLIST" +
 	bash -c "python3 verify_ehc.py --certs-from $CERTS --save-certs tmp-$TRUSTLIST && mv -f tmp-$TRUSTLIST $TRUSTLIST && echo done" &
 fi
 
-LOGFILE="$(date +%Y%m%d%H%M)"
-if compgen -G "*.csv" > /dev/null ; then
-	ALLOW=$(ls *.csv | zenity --list --title="Teilnehmerliste" --column="Datei" )
+# Wähle Teilnehmerliste aus (oder `Abbrechen` für keine)
+CONTACTLOGFILE="$(date +%Y%m%d%H%M)"
+if [[ -d "$ATTENDEEDIR" ]] && compgen -G $ATTENDEEDIR/*.csv > /dev/null ; then
+	ALLOW=$(cd "$ATTENDEEDIR" && ls *.csv | zenity --list --title="Teilnehmerliste" --column="Datei" )
 	if [[ -n "$ALLOW" ]] ; then
-		PARAMS+=" -a $ALLOW"
-		LOGFILE="${ALLOW%.csv}-${LOGFILE}"
+		PARAMS+=" -a $ATTENDEEDIR/$ALLOW"
+		CONTACTLOGFILE="${ALLOW%.csv}-${CONTACTLOGFILE}"
 	fi
 fi
-python3 webcam.py ${PARAMS} -t $TRUSTLIST -l ${LOGFILE}.log -v >${LOGFILE}.err 2>&1
+
+mkdir -p ${CONTACTDIR}
+# Lösche alte Kontaktlisten
+MAXDATE=$(date --date="$CONTACTLIST_MAX_AGE ago" +%Y%m%d%H%M)
+for f in $CONTACTDIR/* ; do
+	if [[ $f =~ ([0-9]{12})\. && ${BASH_REMATCH[1]} -le $MAXDATE ]] ; then
+		rm -f $f
+	fi
+done
+
+# Starte Webcamanwendung (Beenden mit `q`)
+python3 webcam.py ${PARAMS} -t $TRUSTLIST -l "${CONTACTDIR}/${CONTACTLOGFILE}.log" >/dev/null 2>&1
+
+

webcam.py

@@ -17,22 +17,29 @@ import traceback
 from playsound import playsound
 
 parser = argparse.ArgumentParser(description='CovPass Check via Webcam (PoC)')
-parser.add_argument('-v', '--verbose', action='store_true', help='verbose output')
-parser.add_argument('--skip-verification', action='store_true', help='do not verify certificate (for debugging)')
-parser.add_argument('--skip-validation', action='store_true', help='do not validate payload (for debugging)')
-parser.add_argument('--skip-uniquecheck', action='store_true', help='do not check if certificate owner is unique (for debugging)')
-parser.add_argument('-d', '--device', type=int,help='webcam device number (/dev/videoX)', default=0)
-parser.add_argument('-r', '--resolution', help='webcam resolution (WxH)', default='1280x720')
+parser.add_argument('-a', '--allowed', type=argparse.FileType('r'), help='list of allowed names')
 parser.add_argument('-c', '--count', type=int, help='initial count value', default=0)
-parser.add_argument('-s', '--sound', action='store_true', help='accoustic notification after each new (!) scan')
+parser.add_argument('-d', '--device', type=int,help='webcam device number (/dev/videoX)', default=0)
 parser.add_argument('-f', '--fullscreen', action='store_true', help='start in full screen')
 parser.add_argument('-F', '--freeze', type=float, help='seconds to freeze image after each new (!) scan', default=0.7)
-parser.add_argument('-t', '--trustlist', type=argparse.FileType('rb'), help='use given trustlist (json) instead of downloading new one')
-parser.add_argument('-a', '--allowed', type=argparse.FileType('r'), help='list of allowed names')
 parser.add_argument('-l', '--log', type=argparse.FileType('a'), help='access log file', default=sys.stderr)
+parser.add_argument('-m', '--mirror', action='store_true', help='mirror webcam (flip horizontal)')
+parser.add_argument('-r', '--resolution', help='webcam resolution (WxH)', default='1280x720')
+parser.add_argument('-s', '--sound', action='store_true', help='accoustic notification after each new (!) scan')
+parser.add_argument('-t', '--trustlist', type=argparse.FileType('rb'), help='use given trustlist (json) instead of downloading new one')
 parser.add_argument('-w', '--window', help='window name', default='CovPass Check')
+
+debug = parser.add_argument_group('debug', description='These options are only for debugging & testing and should not be used in production!')
+debug.add_argument('-q', '--qrcode', nargs='*', help='Manually process contents of given QR code(s)')
+debug.add_argument('-Q', '--qrfile', type=argparse.FileType('r'), help='Manually process file containing contents of QR code(s)')
+debug.add_argument('-v', '--verbose', action='store_true', help='verbose output')
+debug.add_argument('--skip-verification', action='store_true', help='do not verify certificate')
+debug.add_argument('--skip-validation', action='store_true', help='do not validate payload')
+debug.add_argument('--skip-uniquecheck', action='store_true', help='do not check if certificate owner is unique')
+
 args = parser.parse_args()
 
+
 # Webcam 
 cap = cv2.VideoCapture(args.device)
 w,h = args.resolution.split('x')
@@ -47,32 +54,25 @@ if args.fullscreen:
 else:
     cv2.namedWindow(args.window, cv2.WINDOW_NORMAL)
 
-qrcodes={}
-process=[]
-uniqusers=[]
-validusers=args.count
-EPOCH = datetime(1970, 1, 1)
-
-certs_table: Dict[str, CertList] = {}
 certs: Optional[CertList] = None
 if args.trustlist:
     certs = load_hack_certs_json(args.trustlist.read(), args.trustlist.name)
 else:
     print("Downloading...")
+    certs_table: Dict[str, CertList] = {}
     certs = download_ehc_certs(['DE', 'AT', 'SE', 'GB', 'NL'], certs_table);
 if not certs:
     print("empty trustlist!")
     sys.exit(1)
 
-print("Ready...")
-
-normalize_mapping = [ ('Ä', 'Ae'), ('Ö', 'Oe'), ('Ü', 'Ue'), ('ä', 'ae'), ('ö', 'oe'), ('ü', 'ue'), ('ß', 'ss') ]
 def normalize(s):
-    for f, t in normalize_mapping:
+    for f, t in [ ('Ä', 'Ae'), ('Ö', 'Oe'), ('Ü', 'Ue'), ('ä', 'ae'), ('ö', 'oe'), ('ü', 'ue'), ('ß', 'ss') ]:
        s = s.replace(f, t)
     return ''.join(c for c in unicodedata.normalize('NFD', s) if unicodedata.category(c) != 'Mn').strip()
 
+EPOCH = datetime(1970, 1, 1)
 def verify(ehc_msg, ehc_payload):
+    global certs, EPOCH
     try:
         issued_at = EPOCH + timedelta(seconds=ehc_payload[6])
 
@@ -92,7 +92,7 @@ def validate(data):
     # vaccinated
     if 'v' in data:
         # https://github.com/ehn-dcc-development/ehn-dcc-valuesets/blob/release/2.1.0/vaccine-medicinal-product.json
-        allowed_products = [ 'EU/1/20/1507', 'EU/1/20/1525', 'EU/1/20/1528', 'EU/1/20/1529' ]
+        allowed_products = [ 'EU/1/20/1507', 'EU/1/20/1525', 'EU/1/20/1528', 'EU/1/21/1529' ]
         dose_num = int(data['v'][0]['dn'])
         series_dose = int(data['v'][0]['sd'])
         date = datetime.strptime(data['v'][0]['dt'], '%Y-%m-%d')
@@ -182,6 +182,62 @@ def validate(data):
 
     return False
 
+uniqusers = []
+validusers = args.count
+def process(ehc_code):
+    global args, validusers, uniqusers, allowed
+    if args.verbose:
+        print("New qr code", ehc_code)
+    result = { 'valid': False }
+    try:
+        ehc_msg = decode_ehc(ehc_code)
+        ehc_payload = cbor2.loads(ehc_msg.payload)
+        payload_data = ehc_payload[-260][1]
+        if args.verbose:
+            print("Payload ", payload_data)
+        gn = normalize(payload_data['nam']['gn'])
+        fn = normalize(payload_data['nam']['fn'])
+        result['name'] = gn + ' ' + fn
+
+        # check if unique
+        if args.skip_uniquecheck:
+            result['unique'] = True
+        else:
+            uid = payload_data['nam']['fnt'] + '<<<' + payload_data['nam']['gnt']  + '<<<<' + payload_data['dob']
+            result['uid'] = uid
+            if uid in uniqusers:
+                print(uid, "not unique!")
+                result['unique'] = False
+            else:
+                uniqusers.append(uid)
+                result['unique'] = True
+
+        # verify certificate and validate payload
+        if (args.skip_verification or verify(ehc_msg, ehc_payload)) and (args.skip_validation or validate(payload_data)):
+            result['valid'] = True
+
+            # check if allowed
+            if len(allowed) > 0:
+                result['allowed'] = False
+                gnt = payload_data['nam']['gnt'].replace('<',' ')
+                fnt = payload_data['nam']['fnt'].replace('<',' ')
+                for a in allowed:
+                    if (a[0] in fn and a[1] in gn) or (a[0].upper() in fnt and a[1].upper() in gnt):
+                        result['allowed'] = True
+                        break
+
+            if len(allowed) == 0 or result['allowed']:
+                validusers = validusers + 1
+                note=''
+            else:
+                note='(not in list)'
+            print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), result['name'], note, file=args.log)
+    except:
+        if not 'name' in result:
+            result['name'] = '(Invalid EHC QR)'
+        print(traceback.format_exc())
+    return result
+
 class Color:
     RED = (41,20,141)
     GREEN = (119,155,0)
@@ -218,6 +274,7 @@ def highlight_ehc(frame, obj, qrcode):
      highlight_object(frame, obj, qrcode['name'], col)
 
 def wait(millis = 1):
+    global cv2
     if millis > 0:
         key = cv2.waitKey(millis)
         if key & 0xFF == ord('q'):
@@ -233,9 +290,24 @@ if args.allowed:
 if not cap.isOpened():
     cap.open(WEBCAM_DEVICE)
 
+qrcodes = {}
+if args.qrcode:
+    for ehc_code in args.qrcode:
+        qrcodes[ehc_code] = process(ehc_code)
+if args.qrfile:
+    for ehc_code in args.qrfile:
+        qrcodes[ehc_code] = process(ehc_code)
+
+if args.verbose:
+    print("Ready...")
+
+process_queue = []
 while cap.isOpened():
     # Capture frame-by-frame
     ret, frame = cap.read()
+    # mirror
+    if args.mirror:
+        frame = cv2.flip(frame, 1)
     # Our operations on the frame come here
     im = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
     objs = pyzbar.decode(im)
@@ -248,8 +320,8 @@ while cap.isOpened():
             if ehc_code in qrcodes:
                 highlight_ehc(frame, obj, qrcodes[ehc_code])
             else:
-                # Add to process list
-                process.append((obj, ehc_code))
+                # Add to process queue
+                process_queue.append((obj, ehc_code))
                 highlight_object(frame, obj, '', Color.BLUE)
 
     # Full counter
@@ -258,56 +330,8 @@ while cap.isOpened():
     cv2.imshow(args.window, frame)
 
     # Check unprocessed qr codes
-    for obj, ehc_code in process:
-        if args.verbose:
-            print("New qr code", ehc_code)
-        qrcodes[ehc_code] = { 'valid': False }
-        try:
-            ehc_msg = decode_ehc(ehc_code)
-            ehc_payload = cbor2.loads(ehc_msg.payload)
-            payload_data = ehc_payload[-260][1]
-            if args.verbose:
-                print("Payload ", payload_data)
-            gn = normalize(payload_data['nam']['gn'])
-            fn = normalize(payload_data['nam']['fn'])
-            qrcodes[ehc_code]['name'] = gn + ' ' + fn
-
-            # check if unique
-            if args.skip_uniquecheck:
-                qrcodes[ehc_code]['unique'] = True
-            else:
-                uid = payload_data['nam']['fnt'] + '<<<' + payload_data['nam']['gnt']  + '<<<<' + payload_data['dob']
-                qrcodes[ehc_code]['uid'] = uid
-                if uid in uniqusers:
-                    print(uid, "not unique!")
-                    qrcodes[ehc_code]['unique'] = False
-                else:
-                    uniqusers.append(uid)
-                    qrcodes[ehc_code]['unique'] = True
-
-            # verify certificate and validate payload
-            if (args.skip_verification or verify(ehc_msg, ehc_payload)) and (args.skip_validation or validate(payload_data)):
-                qrcodes[ehc_code]['valid'] = True
-
-                # check if allowed
-                if len(allowed) > 0:
-                    qrcodes[ehc_code]['allowed'] = False
-                    gnt = payload_data['nam']['gnt'].replace('<',' ')
-                    fnt = payload_data['nam']['fnt'].replace('<',' ')
-                    for a in allowed:
-                        if (fn in a[0] and gn in a[1]) or (fnt in a[0].upper() and gnt in a[1].upper()):
-                            qrcodes[ehc_code]['allowed'] = True
-                            break
-
-                if len(allowed) == 0 or qrcodes[ehc_code]['allowed']:
-                    validusers = validusers + 1
-                    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), qrcodes[ehc_code]['name'],file=args.log)
-                else:
-                    print(qrcodes[ehc_code]['name'], 'is not in list!')
-        except:
-            if not 'name' in qrcodes[ehc_code]:
-                qrcodes[ehc_code]['name'] = '(Invalid EHC QR)'
-            print(traceback.format_exc())
+    for obj, ehc_code in process_queue:
+        qrcodes[ehc_code] = process(ehc_code)
 
         # Graphical highlight of code
         highlight_ehc(frame, obj, qrcodes[ehc_code])
@@ -320,7 +344,7 @@ while cap.isOpened():
         # Freeze output (for a short time)
         wait(int(args.freeze * 1000))
 
-    process.clear()
+    process_queue.clear()
     wait()
 
 # When everything done, release the capture