genchallange check has been added for Authorization (#135)

* The genchallange check has been added for Authorization

iso15118/secc/comm_session_handler.py

@@ -92,7 +92,7 @@ def __init__(
         self.selected_auth_option: Optional[AuthEnum] = None
         # The generated challenge sent in PaymentDetailsRes. Its copy is expected in
         # AuthorizationReq (applies to Plug & Charge identification mode only)
-        self.gen_challenge: bytes = bytes(0)
+        self.gen_challenge: Optional[bytes] = None
         # In ISO 15118-2, the EVCCID is the MAC address, given as bytes.
         # In ISO 15118-20, the EVCCID is like a VIN number, given as str.
         self.evcc_id: Union[bytes, str, None] = None

iso15118/secc/states/iso15118_2_states.py

@@ -914,10 +914,10 @@ async def process_message(
                 AuthorizationStatus.ACCEPTED,
                 AuthorizationStatus.ONGOING,
             ]:
-
+                self.comm_session.gen_challenge = get_random_bytes(16)
                 payment_details_res = PaymentDetailsRes(
                     response_code=ResponseCode.OK,
-                    gen_challenge=get_random_bytes(16),
+                    gen_challenge=self.comm_session.gen_challenge,
                     evse_timestamp=time.time(),
                 )
 
@@ -1017,6 +1017,18 @@ async def process_message(
 
         authorization_req: AuthorizationReq = msg.body.authorization_req
 
+        # [V2G2-475] The message 'AuthorizationRes' shall contain the ResponseCode
+        # 'FAILED_ChallengeInvalid' if the challenge response contained in the
+        # AuthorizationReq message in attribute GenChallenge is not valid versus
+        # the provided GenChallenge in PaymentDetailsRes.
+        if authorization_req.gen_challenge != self.comm_session.gen_challenge:
+            self.stop_state_machine(
+                "[V2G2-475] GenChallenge is not the same in PaymentDetailsRes",
+                message,
+                ResponseCode.FAILED_CHALLENGE_INVALID,
+            )
+            return
+
         if self.comm_session.selected_auth_option == AuthEnum.PNC_V2:
             if not self.comm_session.contract_cert_chain:
                 self.stop_state_machine(

tests/secc/states/test_iso15118_2_states.py

@@ -26,6 +26,7 @@
 )
 from iso15118.shared.messages.iso15118_2.body import ResponseCode
 from iso15118.shared.messages.iso15118_2.datatypes import ACEVSEStatus, CertificateChain
+from iso15118.shared.security import get_random_bytes
 from tests.secc.states.test_messages import (
     get_charge_parameter_discovery_req_message_departure_time_one_hour,
     get_charge_parameter_discovery_req_message_no_departure_time,
@@ -187,6 +188,7 @@ async def test_authorization_next_state_on_authorization_request(
         #      `get_dummy_v2g_message_authorization_req`
         self.comm_session.contract_cert_chain = Mock()
         self.comm_session.emaid = "dummy"
+        self.comm_session.gen_challenge = None
         authorization = Authorization(self.comm_session)
         authorization.signature_verified_once = True
         await authorization.process_message(
@@ -202,6 +204,40 @@ async def test_authorization_next_state_on_authorization_request(
             == expected_evse_processing
         )
 
+    async def test_authorization_req_gen_challenge_invalid(self):
+        self.comm_session.writer = Mock()
+        self.comm_session.writer.get_extra_info = Mock()
+
+        self.comm_session.gen_challenge = get_random_bytes(16)
+        id = "aReq"
+        gen_challenge = get_random_bytes(16)
+        authorization = Authorization(self.comm_session)
+
+        await authorization.process_message(
+            message=get_dummy_v2g_message_authorization_req(id, gen_challenge)
+        )
+        assert authorization.next_state == Terminate
+        assert (
+            authorization.message.body.authorization_res.response_code
+            == ResponseCode.FAILED_CHALLENGE_INVALID
+        )
+
+    async def test_authorization_req_gen_challenge_valid(self):
+        self.comm_session.writer = Mock()
+        self.comm_session.writer.get_extra_info = Mock()
+        self.comm_session.selected_auth_option = AuthEnum.PNC_V2
+        self.comm_session.gen_challenge = get_random_bytes(16)
+        id = "aReq"
+        gen_challenge = self.comm_session.gen_challenge
+        self.comm_session.contract_cert_chain = Mock()
+        self.comm_session.emaid = "dummy"
+        authorization = Authorization(self.comm_session)
+        authorization.signature_verified_once = True
+        await authorization.process_message(
+            message=get_dummy_v2g_message_authorization_req(id, gen_challenge)
+        )
+        assert authorization.next_state == ChargeParameterDiscovery
+
     async def test_charge_parameter_discovery_res_v2g2_303(self):
         # V2G2-303 : Sum of individual time intervals shall match the period of time
         # indicated by the EVCC.

tests/secc/states/test_messages.py

@@ -1,4 +1,4 @@
-from typing import List
+from typing import List, Optional
 
 from iso15118.shared.messages.datatypes import (
     PVEAmount,
@@ -118,11 +118,15 @@ def get_dummy_v2g_message_session_stop_req():
     )
 
 
-def get_dummy_v2g_message_authorization_req():
+def get_dummy_v2g_message_authorization_req(
+    id: Optional[str] = None, gen_challenge: Optional[bytes] = None
+):
     # The AuthorizationReq is empty, unless it is following a PaymentDetailsRes
     # message, in which case it must send back the generated challenge.
-    authorization_req = AuthorizationReq()
-
+    if gen_challenge:
+        authorization_req = AuthorizationReq(id=id, gen_challenge=gen_challenge)
+    else:
+        authorization_req = AuthorizationReq()
     return V2GMessage(
         header=MessageHeader(session_id=MOCK_SESSION_ID),
         body=Body(authorization_req=authorization_req),