ACC environment integrated

Author: SchulzeStTSI

tests/test_qrqualitycheck.py

@@ -1,12 +1,15 @@
+import json
+from base64 import b64decode
+import base64
 from os import path
 from pathlib import Path
 from io import BytesIO
 from json import load
-from PIL.Image import open as image_open
+from PIL.Image import NONE, open as image_open
 from base45 import b45decode
 from cbor2 import loads, CBORTag
-from cose.algorithms import Es256, Ps256
-from cose.headers import KID
+from cose.algorithms import Es256, Ps256, Sha256
+from cose.headers import Algorithm, KID
 from cose.keys import CoseKey
 from cose.keys.curves import P256
 from cose.keys.keyops import VerifyOp
@@ -24,19 +27,72 @@
 from zlib import decompress
 from cbor2 import loads, CBORTag
 from datetime import date, datetime, timezone
+import requests
+from filecache import HOUR, MINUTE, filecache
+from json import load
+from cryptography.hazmat.primitives.hashes import SHA256
+from cryptography.hazmat.primitives.asymmetric import ec, rsa
+from cryptography.utils import int_to_bytes
+
 
 TIMESTAMP_ISO8601_EXTENDED = "%Y-%m-%dT%H:%M:%S.%fZ"
 CONFIG_ERROR = 'CONFIG_ERROR'
+ACC_KID_LIST = 'https://dgca-verifier-service-eu-acc.cfapps.eu10.hana.ondemand.com/signercertificateStatus'
+ACC_CERT_LIST= 'https://dgca-verifier-service-eu-acc.cfapps.eu10.hana.ondemand.com/signercertificateUpdate'
 
 def pytest_generate_tests(metafunc):
  if "config_env" in metafunc.fixturenames:
   country_code = metafunc.config.getoption("country_code")
-  # file_name = metafunc.config.getoption("file_name")
-   # print(country_code, file_name)
+  file_name = metafunc.config.getoption("file_name")
+  print(country_code, file_name)
   test_dir = path.dirname(path.dirname(path.abspath(__file__)))
   test_files = glob(str(Path(test_dir, country_code, "*.png")), recursive=True)
   metafunc.parametrize("config_env", test_files, indirect=True)
 
+def getKidList():
+    response= requests.get(ACC_KID_LIST)
+    if not response.ok: 
+        fail("KID List not reachable")
+    kidlist = dict()
+    for x in json.loads(response.text):
+        kidlist[x]=''
+    return kidlist
+      
+
+
+def getCertificates(kidlist):
+    resume_token = 0
+    abort = False
+    while not abort:
+        if resume_token == 0:
+            response= requests.get(ACC_CERT_LIST)
+        else :
+            headers = {"x-resume-token":resume_token}
+            response= requests.get(ACC_CERT_LIST,headers=headers)
+
+        if not response.ok: 
+            fail("Certificate List not reachable")
+        bytes = Sha256.compute_hash(base64.b64decode(response.text))
+
+        kid=base64.b64encode(bytes[0:8]).decode("ascii")
+
+        if kid in kidlist:
+            kidlist[kid] = bytes
+
+        if "x-resume-token" in response.headers:
+            resume_token = response.headers["x-resume-token"]
+        else: 
+            abort= True       
+    return kidlist
+    
+
+@filecache(HOUR)
+def downloadCertificates():
+    kidlist = getKidList()
+    kidlist = getCertificates(kidlist)
+    return kidlist
+
+
 @fixture
 def config_env(request):
     # noinspection PyBroadException
@@ -46,6 +102,7 @@ def config_env(request):
     except Exception:
         return {CONFIG_ERROR: format_exc()}
 
+
 def _readobject(png):
     file = open(png,mode='rb')
     # read all lines at once
@@ -84,6 +141,9 @@ def _checkTags(cose):
         fail(f'QR Code not tagged as Sign1 Message. Tagged with {firstbyte} ({type})')
 
 def test_issuer_quality(config_env: Dict):
+    
+    kidlist = downloadCertificates()
+    
     _PREFIX=config_env
 
     if(not _checkPrefix(_PREFIX)) :
@@ -99,6 +159,22 @@ def test_issuer_quality(config_env: Dict):
 
     _CBOR= Sign1Message.decode(_COSE)
 
+    alg=_CBOR.phdr[Algorithm]
+
+    if not alg in ["Es256","Ps256"] : 
+        fail("Wrong Algorithm used")
+
+    alg=_CBOR.uhdr[Algorithm]  
+    
+    if not alg == NONE: 
+        fail("Algorithm must be in Protected header")  
+
+    print(_CBOR)
 
+    kid = base64.b64encode(_CBOR.phdr[KID]).decode("ascii")
 
+    if not kid in kidlist: 
+        fail("KID exist not on acceptance environment")
 
+    
+