diff --git a/bankid/__init__.py b/bankid/__init__.py index c80df4a..04626e9 100644 --- a/bankid/__init__.py +++ b/bankid/__init__.py @@ -18,16 +18,15 @@ """ -from . import exceptions -from .__version__ import __version__, version -from .certutils import create_bankid_test_server_cert_and_key -from .jsonclient import AsyncBankIDJSONClient, BankIDJSONClient -from .jsonclient6 import BankIDJSONClient6 +from bankid import exceptions +from bankid.__version__ import __version__, version +from bankid.certutils import create_bankid_test_server_cert_and_key +from bankid.syncclient import BankIdClient +from bankid.asyncclient import BankIdAsyncClient __all__ = [ - "BankIDJSONClient", - "AsyncBankIDJSONClient", - "BankIDJSONClient6", + "BankIdClient", + "BankIdAsyncClient", "exceptions", "create_bankid_test_server_cert_and_key", "__version__", diff --git a/bankid/__version__.py b/bankid/__version__.py index f90f7c4..e44bb76 100644 --- a/bankid/__version__.py +++ b/bankid/__version__.py @@ -3,5 +3,5 @@ Version info """ -__version__ = "0.15.0" +__version__ = "1.0.0a1" version = __version__ # backwards compatibility name diff --git a/bankid/asyncclient.py b/bankid/asyncclient.py new file mode 100644 index 0000000..b5ac6b4 --- /dev/null +++ b/bankid/asyncclient.py @@ -0,0 +1,239 @@ +from typing import Optional, Tuple, Dict, Any, Awaitable + +import httpx + +from bankid.base import BankIDClientBaseclass +from bankid.exceptions import get_json_error_class + + +class BankIdAsyncClient(BankIDClientBaseclass): + """The asynchronous client to use for communicating with BankID servers via the v6 API. + + :param certificates: Tuple of string paths to the certificate to use and + the key to sign with. + :type certificates: tuple + :param test_server: Use the test server for authenticating and signing. + :type test_server: bool + :param request_timeout: Timeout for BankID requests. + :type request_timeout: int + + """ + + def __init__(self, certificates: Tuple[str, str], test_server: bool = False, request_timeout: Optional[int] = None): + super().__init__(certificates, test_server, request_timeout) + + kwargs = { + "cert": self.certs, + "headers": {"Content-Type": "application/json"}, + "verify": self.verify_cert, + } + if request_timeout: + kwargs["timeout"] = request_timeout + self.client = httpx.AsyncClient(**kwargs) + + async def authenticate( + self, + end_user_ip: str, + requirement: Dict[str, Any] = None, + user_visible_data: str = None, + user_non_visible_data: str = None, + user_visible_data_format: str = None, + ) -> Dict[str, str]: + """Request an authentication order. The :py:meth:`collect` method + is used to query the status of the order. + + Example data returned: + + .. code-block:: json + + { + "orderRef": "ee3421ea-2096-4000-8130-82648efe0927", + "autoStartToken": "e8df5c3c-c67b-4a01-bfe5-fefeab760beb", + "qrStartToken": "01f94e28-857f-4d8a-bf8e-6c5a24466658", + "qrStartSecret": "b4214886-3b5b-46ab-bc08-6862fddc0e06" + } + + :param end_user_ip: The user IP address as seen by RP. String. IPv4 and IPv6 is allowed. + :type end_user_ip: str + :param requirement: Requirements on how the auth order must be performed. + See the section `Requirements `_ for more details. + :type requirement: dict + :param user_visible_data: Text displayed to the user during authentication with BankID, + with the purpose of providing context for the authentication and to enable users + to detect identification errors and averting fraud attempts. + :type user_visible_data: str + :param user_non_visible_data: Data is not displayed to the user. + :type user_non_visible_data: str + :param user_visible_data_format: If present, and set to “simpleMarkdownV1”, + this parameter indicates that userVisibleData holds formatting characters which + potentially make for a more pleasant user experience. + :type user_visible_data_format: str + :return: The order response. + :rtype: dict + :raises BankIDError: raises a subclass of this error + when error has been returned from server. + + """ + data = self._create_payload( + end_user_ip, + requirement=requirement, + user_visible_data=user_visible_data, + user_non_visible_data=user_non_visible_data, + user_visible_data_format=user_visible_data_format, + ) + + response = await self.client.post(self._auth_endpoint, json=data) + + if response.status_code == 200: + return response.json() + else: + raise get_json_error_class(response) + + async def sign( + self, + end_user_ip, + requirement: Dict[str, Any] = None, + user_visible_data: str = None, + user_non_visible_data: str = None, + user_visible_data_format: str = None, + ) -> Dict[str, str]: + """Request a signing order. The :py:meth:`collect` method + is used to query the status of the order. + + Example data returned: + + .. code-block:: json + + { + "orderRef": "ee3421ea-2096-4000-8130-82648efe0927", + "autoStartToken": "e8df5c3c-c67b-4a01-bfe5-fefeab760beb", + "qrStartToken": "01f94e28-857f-4d8a-bf8e-6c5a24466658", + "qrStartSecret": "b4214886-3b5b-46ab-bc08-6862fddc0e06" + } + + :param end_user_ip: The user IP address as seen by RP. String. IPv4 and IPv6 is allowed. + :type end_user_ip: str + :param requirement: Requirements on how the sign order must be performed. + See the section `Requirements `_ for more details. + :type requirement: dict + :param user_visible_data: Text to be displayed to the user. + :type user_visible_data: str + :param user_non_visible_data: Data is not displayed to the user. + :type user_non_visible_data: str + :param user_visible_data_format: If present, and set to “simpleMarkdownV1”, + this parameter indicates that userVisibleData holds formatting characters which + potentially make for a more pleasant user experience. + :type user_visible_data_format: str + :return: The order response. + :rtype: dict + :raises BankIDError: raises a subclass of this error + when error has been returned from server. + + """ + data = self._create_payload( + end_user_ip, + requirement=requirement, + user_visible_data=user_visible_data, + user_non_visible_data=user_non_visible_data, + user_visible_data_format=user_visible_data_format, + ) + + response = await self.client.post(self._sign_endpoint, json=data) + + if response.status_code == 200: + return response.json() + else: + raise get_json_error_class(response) + + async def collect(self, order_ref: str) -> dict: + """Collects the result of a sign or auth order using the + ``orderRef`` as reference. + + RP should keep on calling collect every two seconds if status is pending. + RP must abort if status indicates failed. The user identity is returned + when complete. + + Example collect results returned while authentication or signing is + still pending: + + .. code-block:: json + + { + "orderRef":"131daac9-16c6-4618-beb0-365768f37288", + "status":"pending", + "hintCode":"userSign" + } + + Example collect result when authentication or signing has failed: + + .. code-block:: json + + { + "orderRef":"131daac9-16c6-4618-beb0-365768f37288", + "status":"failed", + "hintCode":"userCancel" + } + + Example collect result when authentication or signing is successful + and completed: + + .. code-block:: json + + { + "orderRef": "131daac9-16c6-4618-beb0-365768f37288", + "status": "complete", + "completionData": { + "user": { + "personalNumber": "190000000000", + "name": "Karl Karlsson", + "givenName": "Karl", + "surname": "Karlsson" + }, + "device": { + "ipAddress": "192.168.0.1" + }, + "bankIdIssueDate": "2020-02-01", + "signature": "", + "ocspResponse": "" + } + } + + See `BankID Integration Guide `_ + for more details about how to inform end user of the current status, + whether it is pending, failed or completed. + + :param order_ref: The ``orderRef`` UUID returned from auth or sign. + :type order_ref: str + :return: The CollectResponse parsed to a dictionary. + :rtype: dict + :raises BankIDError: raises a subclass of this error + when error has been returned from server. + + """ + response = await self.client.post(self._collect_endpoint, json={"orderRef": order_ref}) + + if response.status_code == 200: + return response.json() + else: + raise get_json_error_class(response) + + async def cancel(self, order_ref: str) -> bool: + """Cancels an ongoing sign or auth order. + + This is typically used if the user cancels the order + in your service or app. + + :param order_ref: The UUID string specifying which order to cancel. + :type order_ref: str + :return: Boolean regarding success of cancellation. + :rtype: bool + :raises BankIDError: raises a subclass of this error + when error has been returned from server. + + """ + response = await self.client.post(self._cancel_endpoint, json={"orderRef": order_ref}) + + if response.status_code == 200: + return response.json() == {} + else: + raise get_json_error_class(response) diff --git a/bankid/base.py b/bankid/base.py new file mode 100644 index 0000000..349e133 --- /dev/null +++ b/bankid/base.py @@ -0,0 +1,78 @@ +import base64 +from datetime import datetime +import hashlib +import hmac +from math import floor +import time +from typing import Tuple, Optional, Dict, Any +from urllib.parse import urljoin + +from bankid.certutils import resolve_cert_path + + +class BankIDClientBaseclass: + """Baseclass for BankID clients.""" + + def __init__( + self, + certificates: Tuple[str, str], + test_server: bool = False, + request_timeout: Optional[int] = None, + ): + self.certs = certificates + self._request_timeout = request_timeout + + if test_server: + self.api_url = "https://appapi2.test.bankid.com/rp/v6.0/" + self.verify_cert = resolve_cert_path("appapi2.test.bankid.com.pem") + else: + self.api_url = "https://appapi2.bankid.com/rp/v6.0/" + self.verify_cert = resolve_cert_path("appapi2.bankid.com.pem") + + self._auth_endpoint = urljoin(self.api_url, "auth") + self._sign_endpoint = urljoin(self.api_url, "sign") + self._collect_endpoint = urljoin(self.api_url, "collect") + self._cancel_endpoint = urljoin(self.api_url, "cancel") + + self.client = None + + @staticmethod + def _encode_user_data(user_data): + if isinstance(user_data, str): + return base64.b64encode(user_data.encode("utf-8")).decode("ascii") + else: + return base64.b64encode(user_data).decode("ascii") + + @staticmethod + def generate_qr_code_content(qr_start_token: str, start_t: [float, datetime], qr_start_secret: str): + """Given QR start token, time.time() or UTC datetime when initiated authentication call was made and the + QR start secret, calculate the current QR code content to display. + """ + if isinstance(start_t, datetime): + start_t = start_t.timestamp() + elapsed_seconds_since_call = int(floor(time.time() - start_t)) + qr_auth_code = hmac.new( + qr_start_secret.encode(), + msg=str(elapsed_seconds_since_call).encode(), + digestmod=hashlib.sha256, + ).hexdigest() + return f"bankid.{qr_start_token}.{elapsed_seconds_since_call}.{qr_auth_code}" + + def _create_payload( + self, + end_user_ip: str, + requirement: Dict[str, Any] = None, + user_visible_data: str = None, + user_non_visible_data: str = None, + user_visible_data_format: str = None, + ): + data = {"endUserIp": end_user_ip} + if requirement and isinstance(requirement, dict): + data["requirement"] = requirement + if user_visible_data: + data["userVisibleData"] = self._encode_user_data(user_visible_data) + if user_non_visible_data: + data["userNonVisibleData"] = self._encode_user_data(user_non_visible_data) + if user_visible_data_format and user_visible_data_format == "simpleMarkdownV1": + data["userVisibleDataFormat"] = "simpleMarkdownV1" + return data diff --git a/bankid/certutils.py b/bankid/certutils.py index 0551b2d..d443d14 100644 --- a/bankid/certutils.py +++ b/bankid/certutils.py @@ -37,7 +37,9 @@ def create_bankid_test_server_cert_and_key(destination_path: str) -> Tuple[str]: """ if os.getenv("TEST_CERT_FILE"): - certificate, key = split_certificate(os.getenv("TEST_CERT_FILE"), destination_path, password=_TEST_CERT_PASSWORD) + certificate, key = split_certificate( + os.getenv("TEST_CERT_FILE"), destination_path, password=_TEST_CERT_PASSWORD + ) else: # Fetch testP12 certificate path diff --git a/bankid/exceptions.py b/bankid/exceptions.py index 28415e6..2ce9bd9 100644 --- a/bankid/exceptions.py +++ b/bankid/exceptions.py @@ -1,5 +1,6 @@ # -*- coding: utf-8 -*- + def get_json_error_class(response): data = response.json() error_class = _JSON_ERROR_CODE_TO_CLASS.get(data.get("errorCode"), BankIDError) diff --git a/bankid/jsonclient.py b/bankid/jsonclient.py deleted file mode 100644 index d6a3920..0000000 --- a/bankid/jsonclient.py +++ /dev/null @@ -1,380 +0,0 @@ -# -*- coding: utf-8 -*- - -import asyncio -import base64 -from typing import Any, Dict, Optional, Tuple, Union -from urllib import parse as urlparse - -import httpx - -from bankid.certutils import resolve_cert_path -from bankid.exceptions import get_json_error_class - - -def _encode_user_data(user_data: Union[str, bytes]) -> str: - if isinstance(user_data, str): - return base64.b64encode(user_data.encode("utf-8")).decode("ascii") - else: - return base64.b64encode(user_data).decode("ascii") - - -class AsyncBankIDJSONClient: - """ - Asynchronous BankID client. - - :param certificates: Tuple of string paths to the certificate to use and - the key to sign with. - :type certificates: tuple - :param test_server: Use the test server for authenticating and signing. - :type test_server: bool - :param request_timeout: Timeout for BankID requests. - :type request_timeout: int - """ - - def __init__( - self, - certificates: Tuple[str], - test_server: bool = False, - request_timeout: Optional[int] = None, - ): - self.certs = certificates - self._request_timeout = request_timeout - - if test_server: - self.api_url = "https://appapi2.test.bankid.com/rp/v5.1/" - self.verify_cert = resolve_cert_path("appapi2.test.bankid.com.pem") - else: - self.api_url = "https://appapi2.bankid.com/rp/v5.1/" - self.verify_cert = resolve_cert_path("appapi2.bankid.com.pem") - - self._auth_endpoint = urlparse.urljoin(self.api_url, "auth") - self._sign_endpoint = urlparse.urljoin(self.api_url, "sign") - self._collect_endpoint = urlparse.urljoin(self.api_url, "collect") - self._cancel_endpoint = urlparse.urljoin(self.api_url, "cancel") - - self.client = httpx.AsyncClient( - cert=self.certs, - headers={"Content-Type": "application/json"}, - verify=self.verify_cert, - timeout=self._request_timeout, - ) - - def authenticate_payload( - self, - end_user_ip: str, - personal_number: Optional[str] = None, - requirement: Optional[str] = None, - **kwargs, - ) -> Dict[str, Any]: - data = {"endUserIp": end_user_ip} - if personal_number: - data["personalNumber"] = personal_number - if requirement and isinstance(requirement, dict): - data["requirement"] = requirement - # Handling potentially changed optional in-parameters. - data.update(kwargs) - return data - - def sign_payload( - self, - end_user_ip: str, - user_visible_data: str, - personal_number: Optional[str] = None, - requirement: Optional[Dict[str, Any]] = None, - user_non_visible_data: Optional[str] = None, - **kwargs, - ) -> Dict[str, Any]: - data = {"endUserIp": end_user_ip} - if personal_number: - data["personalNumber"] = personal_number - data["userVisibleData"] = _encode_user_data(user_visible_data) - if user_non_visible_data: - data["userNonVisibleData"] = _encode_user_data(user_non_visible_data) - if requirement and isinstance(requirement, dict): - data["requirement"] = requirement - # Handling potentially changed optional in-parameters. - data.update(kwargs) - return data - - async def authenticate( - self, - end_user_ip: str, - personal_number: Optional[str] = None, - requirement: Optional[str] = None, - **kwargs, - ) -> Dict[str, Any]: - """Request an authentication order. The :py:meth:`collect` method - is used to query the status of the order. - - Note that personal number is not needed when authentication is to - be done on the same device, provided that the returned - ``autoStartToken`` is used to open the BankID Client. - - Example data returned: - - .. code-block:: json - - { - "orderRef": "ee3421ea-2096-4000-8130-82648efe0927", - "autoStartToken": "e8df5c3c-c67b-4a01-bfe5-fefeab760beb", - "qrStartToken": "01f94e28-857f-4d8a-bf8e-6c5a24466658", - "qrStartSecret": "b4214886-3b5b-46ab-bc08-6862fddc0e06" - } - - :param end_user_ip: IP address of the user requesting - the authentication. - :type end_user_ip: str - :param personal_number: The Swedish personal number in - format YYYYMMDDXXXX. - :type personal_number: str - :param requirement: An optional dictionary stating how the signature - must be created and verified. See BankID Relying Party Guidelines, - section 13.5 for more details. - :type requirement: dict - :return: The order response. - :rtype: dict - :raises BankIDError: raises a subclass of this error - when error has been returned from server. - """ - response = await self.client.post( - self._auth_endpoint, - json=self.authenticate_payload( - end_user_ip, - personal_number, - requirement, - **kwargs, - ), - ) - if response.status_code == 200: - return response.json() - - raise get_json_error_class(response) - - async def sign( - self, - end_user_ip: str, - user_visible_data: str, - personal_number: Optional[str] = None, - requirement: Optional[Dict[str, Any]] = None, - user_non_visible_data: Optional[str] = None, - **kwargs, - ) -> Dict[str, Any]: - """Request a signing order. The :py:meth:`collect` method - is used to query the status of the order. - - Note that personal number is not needed when signing is to be done - on the same device, provided that the returned ``autoStartToken`` - is used to open the BankID Client. - - Example data returned: - - .. code-block:: json - - { - "orderRef": "ee3421ea-2096-4000-8130-82648efe0927", - "autoStartToken": "e8df5c3c-c67b-4a01-bfe5-fefeab760beb", - "qrStartToken": "01f94e28-857f-4d8a-bf8e-6c5a24466658", - "qrStartSecret": "b4214886-3b5b-46ab-bc08-6862fddc0e06" - } - - :param end_user_ip: IP address of the user requesting - the authentication. - :type end_user_ip: str - :param user_visible_data: The information that the end user - is requested to sign. - :type user_visible_data: str - :param personal_number: The Swedish personal number in - format YYYYMMDDXXXX. - :type personal_number: str - :param requirement: An optional dictionary stating how the signature - must be created and verified. See BankID Relying Party Guidelines, - section 13.5 for more details. - :type requirement: dict - :param user_non_visible_data: Optional information sent with request - that the user never sees. - :type user_non_visible_data: str - :return: The order response. - :rtype: dict - :raises BankIDError: raises a subclass of this error - when error has been returned from server. - """ - response = await self.client.post( - self._sign_endpoint, - json=self.sign_payload( - end_user_ip, - user_visible_data, - personal_number, - requirement, - user_non_visible_data, - **kwargs, - ), - ) - if response.status_code == 200: - return response.json() - - raise get_json_error_class(response) - - async def collect(self, order_ref: str) -> Dict[str, Any]: - """Collects the result of a sign or auth order using the - ``orderRef`` as reference. - - RP should keep on calling collect every two seconds as long as status - indicates pending. RP must abort if status indicates failed. The user - identity is returned when complete. - - Example collect results returned while authentication or signing is - still pending: - - .. code-block:: json - - { - "orderRef":"131daac9-16c6-4618-beb0-365768f37288", - "status":"pending", - "hintCode":"userSign" - } - - Example collect result when authentication or signing has failed: - - .. code-block:: json - - { - "orderRef":"131daac9-16c6-4618-beb0-365768f37288", - "status":"failed", - "hintCode":"userCancel" - } - - Example collect result when authentication or signing is successful - and completed: - - .. code-block:: json - - { - "orderRef":"131daac9-16c6-4618-beb0-365768f37288", - "status":"complete", - "completionData": { - "user": { - "personalNumber":"190000000000", - "name":"Karl Karlsson", - "givenName":"Karl", - "surname":"Karlsson" - }, - "device": { - "ipAddress":"192.168.0.1" - }, - "cert": { - "notBefore":"1502983274000", - "notAfter":"1563549674000" - }, - "signature":"", - "ocspResponse":"" - } - } - - See `BankID Relying Party Guidelines Version: 3.5 `_ - for more details about how to inform end user of the current status, - whether it is pending, failed or completed. - - :param order_ref: The ``orderRef`` UUID returned from auth or sign. - :type order_ref: str - :return: The CollectResponse parsed to a dictionary. - :rtype: dict - :raises BankIDError: raises a subclass of this error - when error has been returned from server. - """ - response = await self.client.post( - self._collect_endpoint, - json=dict( - orderRef=order_ref, - ), - ) - if response.status_code == 200: - return response.json() - - raise get_json_error_class(response) - - async def cancel(self, order_ref: str) -> bool: - """Cancels an ongoing sign or auth order. - - This is typically used if the user cancels the order - in your service or app. - - :param order_ref: The UUID string specifying which order to cancel. - :type order_ref: str - :return: Boolean regarding success of cancellation. - :rtype: bool - :raises BankIDError: raises a subclass of this error - when error has been returned from server. - """ - response = await self.client.post( - self._cancel_endpoint, - json=dict( - orderRef=order_ref, - ), - ) - if response.status_code == 200: - return response.json() == {} - - raise get_json_error_class(response) - - -class BankIDJSONClient(AsyncBankIDJSONClient): - """Synchronous BankID client. - - :param certificates: Tuple of string paths to the certificate to use and - the key to sign with. - :type certificates: tuple - :param test_server: Use the test server for authenticating and signing. - :type test_server: bool - :param request_timeout: Timeout for BankID requests. - :type request_timeout: int - """ - - def __init__( - self, - certificates: Tuple[str], - test_server: bool = False, - request_timeout: Optional[int] = None, - ): - self.loop = asyncio.new_event_loop() - self.async_runner = self.loop.run_until_complete - self.async_client = super() - self.async_client.__init__(certificates, test_server, request_timeout) - - def __del__(self): - self.loop.close() - - def cancel( - self, - order_ref: str, - ) -> Dict[str, Any]: - return self.async_runner( - self.async_client.cancel(order_ref), - ) - - def collect( - self, - order_ref: str, - ) -> Dict[str, Any]: - return self.async_runner( - self.async_client.collect(order_ref), - ) - - def sign( - self, - ip_address: str, - user_visible_data: str, - personal_number: Optional[str] = None, - user_non_visible_data: Optional[str] = None, - ) -> Dict[str, Any]: - return self.async_runner( - self.async_client.sign(ip_address, user_visible_data, personal_number, user_non_visible_data), - ) - - def authenticate( - self, - ip_address: str, - personal_number: Optional[str] = None, - ) -> Dict[str, Any]: - return self.async_runner( - self.async_client.authenticate(ip_address, personal_number), - ) diff --git a/bankid/jsonclient6.py b/bankid/jsonclient6.py deleted file mode 100644 index c82190e..0000000 --- a/bankid/jsonclient6.py +++ /dev/null @@ -1,273 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -:mod:`bankid.jsonclient6` -- BankID JSON Client -============================================== - -Created on 2024-01-18 by mxamin - -""" -import base64 -from urllib import parse as urlparse - -import requests - -from bankid.certutils import resolve_cert_path -from bankid.exceptions import get_json_error_class - - -def _encode_user_data(user_data): - if isinstance(user_data, str): - return base64.b64encode(user_data.encode("utf-8")).decode("ascii") - else: - return base64.b64encode(user_data).decode("ascii") - - -class BankIDJSONClient6(object): - """The client to use for communicating with BankID servers via the v.5 API. - - :param certificates: Tuple of string paths to the certificate to use and - the key to sign with. - :type certificates: tuple - :param test_server: Use the test server for authenticating and signing. - :type test_server: bool - :param request_timeout: Timeout for BankID requests. - :type request_timeout: int - - """ - - def __init__(self, certificates, test_server=False, request_timeout=None): - self.certs = certificates - self._request_timeout = request_timeout - - if test_server: - self.api_url = "https://appapi2.test.bankid.com/rp/v6.0/" - self.verify_cert = resolve_cert_path("appapi2.test.bankid.com.pem") - else: - self.api_url = "https://appapi2.bankid.com/rp/v6.0/" - self.verify_cert = resolve_cert_path("appapi2.bankid.com.pem") - - self.client = requests.Session() - self.client.verify = self.verify_cert - self.client.cert = self.certs - self.client.headers = {"Content-Type": "application/json"} - - self._auth_endpoint = urlparse.urljoin(self.api_url, "auth") - self._sign_endpoint = urlparse.urljoin(self.api_url, "sign") - self._collect_endpoint = urlparse.urljoin(self.api_url, "collect") - self._cancel_endpoint = urlparse.urljoin(self.api_url, "cancel") - - def _post(self, endpoint, *args, **kwargs): - """Internal helper method for adding timeout to requests.""" - return self.client.post( - endpoint, *args, timeout=self._request_timeout, **kwargs - ) - - def authenticate( - self, - end_user_ip, - requirement=None, - user_visible_data=None, - user_non_visible_data=None, - **kwargs - ): - """Request an authentication order. The :py:meth:`collect` method - is used to query the status of the order. - - Example data returned: - - .. code-block:: json - - { - "orderRef": "ee3421ea-2096-4000-8130-82648efe0927", - "autoStartToken": "e8df5c3c-c67b-4a01-bfe5-fefeab760beb", - "qrStartToken": "01f94e28-857f-4d8a-bf8e-6c5a24466658", - "qrStartSecret": "b4214886-3b5b-46ab-bc08-6862fddc0e06" - } - - :param end_user_ip: IP address of the user requesting - the authentication. - :type end_user_ip: str - :param requirement: An optional dictionary stating how the signature - must be created and verified. See BankID Relying Party Integration Guide for v6.0 - for more details. - :type requirement: dict - :param user_visible_data: The information that the end user - is requested to sign. - :type user_visible_data: str - :param user_non_visible_data: Optional information sent with request - that the user never sees. - :type user_non_visible_data: str - :return: The order response. - :rtype: dict - :raises BankIDError: raises a subclass of this error - when error has been returned from server. - - """ - data = {"endUserIp": end_user_ip} - if requirement and isinstance(requirement, dict): - data["requirement"] = requirement - if user_visible_data: - data["userVisibleData"] = _encode_user_data(user_visible_data) - if user_non_visible_data: - data["userNonVisibleData"] = _encode_user_data(user_non_visible_data) - # Handling potentially changed optional in-parameters. - data.update(kwargs) - response = self._post(self._auth_endpoint, json=data) - - if response.status_code == 200: - return response.json() - else: - raise get_json_error_class(response) - - def sign( - self, - end_user_ip, - user_visible_data, - requirement=None, - user_non_visible_data=None, - **kwargs - ): - """Request a signing order. The :py:meth:`collect` method - is used to query the status of the order. - - Example data returned: - - .. code-block:: json - - { - "orderRef": "ee3421ea-2096-4000-8130-82648efe0927", - "autoStartToken": "e8df5c3c-c67b-4a01-bfe5-fefeab760beb", - "qrStartToken": "01f94e28-857f-4d8a-bf8e-6c5a24466658", - "qrStartSecret": "b4214886-3b5b-46ab-bc08-6862fddc0e06" - } - - :param end_user_ip: IP address of the user requesting - the authentication. - :type end_user_ip: str - :param user_visible_data: The information that the end user - is requested to sign. - :type user_visible_data: str - :param requirement: An optional dictionary stating how the signature - must be created and verified. See BankID Relying Party Integration Guide for v6.0 - for more details. - :type requirement: dict - :param user_non_visible_data: Optional information sent with request - that the user never sees. - :type user_non_visible_data: str - :return: The order response. - :rtype: dict - :raises BankIDError: raises a subclass of this error - when error has been returned from server. - - """ - data = {"endUserIp": end_user_ip} - data["userVisibleData"] = _encode_user_data(user_visible_data) - if user_non_visible_data: - data["userNonVisibleData"] = _encode_user_data(user_non_visible_data) - if requirement and isinstance(requirement, dict): - data["requirement"] = requirement - # Handling potentially changed optional in-parameters. - data.update(kwargs) - response = self._post(self._sign_endpoint, json=data) - - if response.status_code == 200: - return response.json() - else: - raise get_json_error_class(response) - - def collect(self, order_ref): - """Collects the result of a sign or auth order using the - ``orderRef`` as reference. - - RP should keep on calling collect every two seconds as long as status - indicates pending. RP must abort if status indicates failed. The user - identity is returned when complete. - - Example collect results returned while authentication or signing is - still pending: - - .. code-block:: json - - { - "orderRef":"131daac9-16c6-4618-beb0-365768f37288", - "status":"pending", - "hintCode":"userSign" - } - - Example collect result when authentication or signing has failed: - - .. code-block:: json - - { - "orderRef":"131daac9-16c6-4618-beb0-365768f37288", - "status":"failed", - "hintCode":"userCancel" - } - - Example collect result when authentication or signing is successful - and completed: - - .. code-block:: json - - { - "orderRef":"131daac9-16c6-4618-beb0-365768f37288", - "status":"complete", - "completionData": { - "user": { - "personalNumber":"190000000000", - "name":"Karl Karlsson", - "givenName":"Karl", - "surname":"Karlsson" - }, - "device": { - "ipAddress":"192.168.0.1" - }, - "cert": { - "notBefore":"1502983274000", - "notAfter":"1563549674000" - }, - "signature":"", - "ocspResponse":"" - } - } - - See `BankID Relying Party Integration Guide `_ - for more details about how to inform end user of the current status, - whether it is pending, failed or completed. - - :param order_ref: The ``orderRef`` UUID returned from auth or sign. - :type order_ref: str - :return: The CollectResponse parsed to a dictionary. - :rtype: dict - :raises BankIDError: raises a subclass of this error - when error has been returned from server. - - """ - response = self._post(self._collect_endpoint, json={"orderRef": order_ref}) - - if response.status_code == 200: - return response.json() - else: - raise get_json_error_class(response) - - def cancel(self, order_ref): - """Cancels an ongoing sign or auth order. - - This is typically used if the user cancels the order - in your service or app. - - :param order_ref: The UUID string specifying which order to cancel. - :type order_ref: str - :return: Boolean regarding success of cancellation. - :rtype: bool - :raises BankIDError: raises a subclass of this error - when error has been returned from server. - - """ - response = self._post(self._cancel_endpoint, json={"orderRef": order_ref}) - - if response.status_code == 200: - return response.json() == {} - else: - raise get_json_error_class(response) diff --git a/bankid/syncclient.py b/bankid/syncclient.py new file mode 100644 index 0000000..67d5fc9 --- /dev/null +++ b/bankid/syncclient.py @@ -0,0 +1,238 @@ +from typing import Optional, Tuple, Dict, Any + +import httpx + +from bankid.base import BankIDClientBaseclass +from bankid.exceptions import get_json_error_class + + +class BankIdClient(BankIDClientBaseclass): + """The synchronous client to use for communicating with BankID servers via the v6 API. + + :param certificates: Tuple of string paths to the certificate to use and + the key to sign with. + :type certificates: tuple + :param test_server: Use the test server for authenticating and signing. + :type test_server: bool + :param request_timeout: Timeout for BankID requests. + :type request_timeout: int + + """ + + def __init__(self, certificates: Tuple[str], test_server: bool = False, request_timeout: Optional[int] = None): + super().__init__(certificates, test_server, request_timeout) + + kwargs = { + "cert": self.certs, + "headers": {"Content-Type": "application/json"}, + "verify": self.verify_cert, + } + if request_timeout: + kwargs["timeout"] = request_timeout + self.client = httpx.Client(**kwargs) + + def authenticate( + self, + end_user_ip: str, + requirement: Dict[str, Any] = None, + user_visible_data: str = None, + user_non_visible_data: str = None, + user_visible_data_format: str = None, + ) -> Dict[str, str]: + """Request an authentication order. The :py:meth:`collect` method + is used to query the status of the order. + + Example data returned: + + .. code-block:: json + + { + "orderRef": "ee3421ea-2096-4000-8130-82648efe0927", + "autoStartToken": "e8df5c3c-c67b-4a01-bfe5-fefeab760beb", + "qrStartToken": "01f94e28-857f-4d8a-bf8e-6c5a24466658", + "qrStartSecret": "b4214886-3b5b-46ab-bc08-6862fddc0e06" + } + + :param end_user_ip: The user IP address as seen by RP. String. IPv4 and IPv6 is allowed. + :type end_user_ip: str + :param requirement: Requirements on how the auth order must be performed. + See the section `Requirements `_ for more details. + :type requirement: dict + :param user_visible_data: Text displayed to the user during authentication with BankID, + with the purpose of providing context for the authentication and to enable users + to detect identification errors and averting fraud attempts. + :type user_visible_data: str + :param user_non_visible_data: Data is not displayed to the user. + :type user_non_visible_data: str + :param user_visible_data_format: If present, and set to “simpleMarkdownV1”, + this parameter indicates that userVisibleData holds formatting characters which + potentially make for a more pleasant user experience. + :type user_visible_data_format: str + :return: The order response. + :rtype: dict + :raises BankIDError: raises a subclass of this error + when error has been returned from server. + + """ + data = self._create_payload( + end_user_ip, + requirement=requirement, + user_visible_data=user_visible_data, + user_non_visible_data=user_non_visible_data, + user_visible_data_format=user_visible_data_format, + ) + + response = self.client.post(self._auth_endpoint, json=data) + + if response.status_code == 200: + return response.json() + else: + raise get_json_error_class(response) + + def sign( + self, + end_user_ip: str, + requirement: Dict[str, Any] = None, + user_visible_data: str = None, + user_non_visible_data: str = None, + user_visible_data_format: str = None, + ) -> Dict[str, str]: + """Request a signing order. The :py:meth:`collect` method + is used to query the status of the order. + + Example data returned: + + .. code-block:: json + + { + "orderRef": "ee3421ea-2096-4000-8130-82648efe0927", + "autoStartToken": "e8df5c3c-c67b-4a01-bfe5-fefeab760beb", + "qrStartToken": "01f94e28-857f-4d8a-bf8e-6c5a24466658", + "qrStartSecret": "b4214886-3b5b-46ab-bc08-6862fddc0e06" + } + + :param end_user_ip: The user IP address as seen by RP. String. IPv4 and IPv6 is allowed. + :type end_user_ip: str + :param requirement: Requirements on how the sign order must be performed. + See the section `Requirements `_ for more details. + :type requirement: dict + :param user_visible_data: Text to be displayed to the user. + :type user_visible_data: str + :param user_non_visible_data: Data is not displayed to the user. + :type user_non_visible_data: str + :param user_visible_data_format: If present, and set to “simpleMarkdownV1”, + this parameter indicates that userVisibleData holds formatting characters which + potentially make for a more pleasant user experience. + :type user_visible_data_format: str + :return: The order response. + :rtype: dict + :raises BankIDError: raises a subclass of this error + when error has been returned from server. + + """ + data = self._create_payload( + end_user_ip, + requirement=requirement, + user_visible_data=user_visible_data, + user_non_visible_data=user_non_visible_data, + user_visible_data_format=user_visible_data_format, + ) + response = self.client.post(self._sign_endpoint, json=data) + + if response.status_code == 200: + return response.json() + else: + raise get_json_error_class(response) + + def collect(self, order_ref: str) -> dict: + """Collects the result of a sign or auth order using the + ``orderRef`` as reference. + + RP should keep on calling collect every two seconds if status is pending. + RP must abort if status indicates failed. The user identity is returned + when complete. + + Example collect results returned while authentication or signing is + still pending: + + .. code-block:: json + + { + "orderRef":"131daac9-16c6-4618-beb0-365768f37288", + "status":"pending", + "hintCode":"userSign" + } + + Example collect result when authentication or signing has failed: + + .. code-block:: json + + { + "orderRef":"131daac9-16c6-4618-beb0-365768f37288", + "status":"failed", + "hintCode":"userCancel" + } + + Example collect result when authentication or signing is successful + and completed: + + .. code-block:: json + + { + "orderRef": "131daac9-16c6-4618-beb0-365768f37288", + "status": "complete", + "completionData": { + "user": { + "personalNumber": "190000000000", + "name": "Karl Karlsson", + "givenName": "Karl", + "surname": "Karlsson" + }, + "device": { + "ipAddress": "192.168.0.1" + }, + "bankIdIssueDate": "2020-02-01", + "signature": "", + "ocspResponse": "" + } + } + + See `BankID Integration Guide `_ + for more details about how to inform end user of the current status, + whether it is pending, failed or completed. + + :param order_ref: The ``orderRef`` UUID returned from auth or sign. + :type order_ref: str + :return: The CollectResponse parsed to a dictionary. + :rtype: dict + :raises BankIDError: raises a subclass of this error + when error has been returned from server. + + """ + response = self.client.post(self._collect_endpoint, json={"orderRef": order_ref}) + + if response.status_code == 200: + return response.json() + else: + raise get_json_error_class(response) + + def cancel(self, order_ref: str) -> bool: + """Cancels an ongoing sign or auth order. + + This is typically used if the user cancels the order + in your service or app. + + :param order_ref: The UUID string specifying which order to cancel. + :type order_ref: str + :return: Boolean regarding success of cancellation. + :rtype: bool + :raises BankIDError: raises a subclass of this error + when error has been returned from server. + + """ + response = self.client.post(self._cancel_endpoint, json={"orderRef": order_ref}) + + if response.status_code == 200: + return response.json() == {} + else: + raise get_json_error_class(response) diff --git a/examples/qrdemo/qrdemo/app.py b/examples/qrdemo/qrdemo/app.py index 54e2446..68dad8a 100644 --- a/examples/qrdemo/qrdemo/app.py +++ b/examples/qrdemo/qrdemo/app.py @@ -7,7 +7,7 @@ from flask import Flask, make_response, render_template, request, jsonify from flask_caching import Cache -from bankid import BankIDJSONClient +from bankid import BankIdClient from bankid.certutils import create_bankid_test_server_cert_and_key USE_TEST_SERVER = True @@ -19,13 +19,13 @@ # Flask app. For this demo it is sufficient to let it reside globally in this file. if USE_TEST_SERVER: cert_paths = create_bankid_test_server_cert_and_key(str(pathlib.Path(__file__).parent)) - client = BankIDJSONClient(cert_paths, test_server=True) + client = BankIdClient(cert_paths, test_server=True) else: # Set your own cert paths for you production certificate and key here. # Note that my recommendation is to get it to work with # test server certs first! cert_paths = ("certificate.pem", "key.pem") - client = BankIDJSONClient(cert_paths, test_server=False) + client = BankIdClient(cert_paths, test_server=False) # Frontend pages @@ -74,8 +74,7 @@ def initiate(): # Make Auth call to BankID. resp = client.authenticate( end_user_ip=request.remote_addr, # Get the IP of the device making the request. - personal_number=pn, - requirement={"tokenStartRequired": True if pn else False}, # Set to True if PN is provided. Recommended. + requirement={"personalNumber": pn}, # Set to True if PN is provided. Recommended. ) # Record when this response was received. This is needed for generating sequential, animated QR codes. resp["start_t"] = time.time() @@ -83,7 +82,7 @@ def initiate(): # multi-instance apps. Using orderRef as key since it is unique and can be sent in a GET URL without problem. cache.set(resp.get("orderRef"), resp, timeout=5 * 60) # Generate the first QR code to display to user. - qr_content_0 = generate_qr_code_content(resp["qrStartToken"], resp["start_t"], resp["qrStartSecret"]) + qr_content_0 = client.generate_qr_code_content(resp["qrStartToken"], resp["start_t"], resp["qrStartSecret"]) return render_template( "qr.html", order_ref=resp["orderRef"], @@ -99,7 +98,7 @@ def get_qr_code(order_ref: str): if x is None: qr_content = "" else: - qr_content = generate_qr_code_content(x["qrStartToken"], x["start_t"], x["qrStartSecret"]) + qr_content = client.generate_qr_code_content(x["qrStartToken"], x["start_t"], x["qrStartSecret"]) response = make_response(qr_content, 200) response.mimetype = "text/plain" return response @@ -123,19 +122,3 @@ def collect(order_ref: str): return response else: return jsonify(collect_response) - - -# Helper methods - - -def generate_qr_code_content(qr_start_token: str, start_t: float, qr_start_secret: str): - """Given QR start token, time.time() when initiated authentication call was made and the - QR start secret, calculate the current QR code content to display. - """ - elapsed_seconds_since_call = int(floor(time.time() - start_t)) - qr_auth_code = hmac.new( - qr_start_secret.encode(), - msg=str(elapsed_seconds_since_call).encode(), - digestmod=hashlib.sha256, - ).hexdigest() - return f"bankid.{qr_start_token}.{elapsed_seconds_since_call}.{qr_auth_code}" diff --git a/requirements.txt b/requirements.txt index 3244f28..5690bb5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,2 @@ -httpx==0.24.1 -importlib-resources==5.12.0 +httpx +importlib-resources>=5.12.0 diff --git a/tests/conftest.py b/tests/conftest.py index ead8e00..0cb3405 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,4 +1,5 @@ import random +from typing import Awaitable import httpx import pytest @@ -7,11 +8,18 @@ from bankid.certs import get_test_cert_and_key +@pytest.fixture() +def ip_address() -> str: + with httpx.Client() as client: + response = client.get("https://httpbin.org/ip") + return response.json()["origin"].split(",")[0] + + @pytest_asyncio.fixture() -async def ip_address(): - client = httpx.AsyncClient() - response = await client.get("https://httpbin.org/ip") - return response.json()["origin"].split(",")[0] +async def ip_address_async() -> str: + async with httpx.AsyncClient() as client: + response = await client.get("https://httpbin.org/ip") + return response.json()["origin"].split(",")[0] @pytest.fixture() diff --git a/tests/test_jsonclient_async.py b/tests/test_asyncclient.py similarity index 53% rename from tests/test_jsonclient_async.py rename to tests/test_asyncclient.py index 0566835..64cc2f8 100644 --- a/tests/test_jsonclient_async.py +++ b/tests/test_asyncclient.py @@ -1,16 +1,30 @@ +""" +:mod:`test_asyncclient` +======================= + +.. module:: test_asyncclient + :platform: Unix, Windows + :synopsis: + +.. moduleauthor:: tiwilliam + +Created on 2023-12-15 + +""" + import uuid import pytest -import bankid +from bankid import BankIdAsyncClient, exceptions @pytest.mark.asyncio -async def test_authentication_and_collect(cert_and_key, ip_address, random_personal_number): +async def test_authentication_and_collect(cert_and_key, ip_address_async): """Authenticate call and then collect with the returned orderRef UUID.""" - c = bankid.AsyncBankIDJSONClient(certificates=cert_and_key, test_server=True) + c = BankIdAsyncClient(certificates=cert_and_key, test_server=True) assert "appapi2.test.bankid.com.pem" in c.verify_cert - out = await c.authenticate(ip_address, random_personal_number) + out = await c.authenticate(ip_address_async) assert isinstance(out, dict) # UUID.__init__ performs the UUID compliance assertion. uuid.UUID(out.get("orderRef"), version=4) @@ -20,14 +34,13 @@ async def test_authentication_and_collect(cert_and_key, ip_address, random_perso @pytest.mark.asyncio -async def test_sign_and_collect(cert_and_key, ip_address, random_personal_number): +async def test_sign_and_collect(cert_and_key, ip_address_async): """Sign call and then collect with the returned orderRef UUID.""" - c = bankid.AsyncBankIDJSONClient(certificates=cert_and_key, test_server=True) + c = BankIdAsyncClient(certificates=cert_and_key, test_server=True) out = await c.sign( - ip_address, - "The data to be signed", - personal_number=random_personal_number, + ip_address_async, + user_visible_data="The data to be signed", user_non_visible_data="Non visible data", ) assert isinstance(out, dict) @@ -40,32 +53,38 @@ async def test_sign_and_collect(cert_and_key, ip_address, random_personal_number @pytest.mark.asyncio async def test_invalid_orderref_raises_error(cert_and_key): - c = bankid.AsyncBankIDJSONClient(certificates=cert_and_key, test_server=True) - with pytest.raises(bankid.exceptions.InvalidParametersError): + c = BankIdAsyncClient(certificates=cert_and_key, test_server=True) + with pytest.raises(exceptions.InvalidParametersError): await c.collect("invalid-uuid") @pytest.mark.asyncio -async def test_already_in_progress_raises_error(cert_and_key, ip_address, random_personal_number): - c = bankid.AsyncBankIDJSONClient(certificates=cert_and_key, test_server=True) - await c.authenticate(ip_address, random_personal_number) - with pytest.raises(bankid.exceptions.AlreadyInProgressError): - await c.authenticate(ip_address, random_personal_number) +async def test_already_in_progress_raises_error(cert_and_key, ip_address_async, random_personal_number): + c = BankIdAsyncClient(certificates=cert_and_key, test_server=True) + await c.authenticate(ip_address_async, requirement={"personalNumber": random_personal_number}) + with pytest.raises(exceptions.AlreadyInProgressError): + await c.authenticate(ip_address_async, requirement={"personalNumber": random_personal_number}) @pytest.mark.asyncio -async def test_already_in_progress_raises_error_2(cert_and_key, ip_address, random_personal_number): - c = bankid.AsyncBankIDJSONClient(certificates=cert_and_key, test_server=True) - await c.sign(ip_address, "Text to sign", random_personal_number) - with pytest.raises(bankid.exceptions.AlreadyInProgressError): - await c.sign(ip_address, "Text to sign", random_personal_number) +async def test_already_in_progress_raises_error_2(cert_and_key, ip_address_async, random_personal_number): + c = BankIdAsyncClient(certificates=cert_and_key, test_server=True) + await c.sign( + ip_address_async, + requirement={"personalNumber": random_personal_number}, + user_visible_data="Text to sign", + ) + with pytest.raises(exceptions.AlreadyInProgressError): + await c.sign( + ip_address_async, requirement={"personalNumber": random_personal_number}, user_visible_data="Text to sign" + ) @pytest.mark.asyncio -async def test_authentication_and_cancel(cert_and_key, ip_address, random_personal_number): +async def test_authentication_and_cancel(cert_and_key, ip_address_async): """Authenticate call and then cancel it""" - c = bankid.AsyncBankIDJSONClient(certificates=cert_and_key, test_server=True) - out = await c.authenticate(ip_address, random_personal_number) + c = BankIdAsyncClient(certificates=cert_and_key, test_server=True) + out = await c.authenticate(ip_address_async) assert isinstance(out, dict) # UUID.__init__ performs the UUID compliance assertion. order_ref = uuid.UUID(out.get("orderRef"), version=4) @@ -74,13 +93,13 @@ async def test_authentication_and_cancel(cert_and_key, ip_address, random_person assert collect_status.get("hintCode") in ("outstandingTransaction", "noClient") success = await c.cancel(str(order_ref)) assert success - with pytest.raises(bankid.exceptions.InvalidParametersError): + with pytest.raises(exceptions.InvalidParametersError): collect_status = await c.collect(out.get("orderRef")) @pytest.mark.asyncio async def test_cancel_with_invalid_uuid(cert_and_key): - c = bankid.AsyncBankIDJSONClient(certificates=cert_and_key, test_server=True) + c = BankIdAsyncClient(certificates=cert_and_key, test_server=True) invalid_order_ref = uuid.uuid4() - with pytest.raises(bankid.exceptions.InvalidParametersError): + with pytest.raises(exceptions.InvalidParametersError): await c.cancel(str(invalid_order_ref)) diff --git a/tests/test_jsonclient.py b/tests/test_jsonclient.py deleted file mode 100644 index be2a43e..0000000 --- a/tests/test_jsonclient.py +++ /dev/null @@ -1,13 +0,0 @@ -import pytest - -import bankid - - -@pytest.mark.parametrize( - "test_server, endpoint", - [(False, "appapi2.bankid.com"), (True, "appapi2.test.bankid.com")], -) -def test_correct_prod_server_urls(cert_and_key, test_server, endpoint): - c = bankid.BankIDJSONClient(certificates=cert_and_key, test_server=test_server) - assert c.api_url == "https://{0}/rp/v5.1/".format(endpoint) - assert "{0}.pem".format(endpoint) in c.verify_cert diff --git a/tests/test_jsonclient6.py b/tests/test_jsonclient6.py deleted file mode 100644 index a6ba8e1..0000000 --- a/tests/test_jsonclient6.py +++ /dev/null @@ -1,157 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -:mod:`test_client` -================== - -.. module:: test_client - :platform: Unix, Windows - :synopsis: - -.. moduleauthor:: mxamin - -Created on 2024-01-18 - -""" - -from __future__ import division -from __future__ import print_function -from __future__ import unicode_literals -from __future__ import absolute_import - -import random -import tempfile -import uuid - -import pytest - -try: - from unittest import mock -except: - import mock - -import bankid - - -def _get_random_personal_number(): - """Simple random Swedish personal number generator.""" - - def _luhn_digit(id_): - """Calculate Luhn control digit for personal number. - - Code adapted from `Faker - `_. - - :param id_: The partial number to calculate checksum of. - :type id_: str - :return: Integer digit in [0, 9]. - :rtype: int - - """ - - def digits_of(n): - return [int(i) for i in str(n)] - - id_ = int(id_) * 10 - digits = digits_of(id_) - checksum = sum(digits[-1::-2]) - for k in digits[-2::-2]: - checksum += sum(digits_of(k * 2)) - checksum %= 10 - - return checksum if checksum == 0 else 10 - checksum - - year = random.randint(1900, 2014) - month = random.randint(1, 12) - day = random.randint(1, 28) - suffix = random.randint(0, 999) - pn = "{0:04d}{1:02d}{2:02d}{3:03d}".format(year, month, day, suffix) - return pn + str(_luhn_digit(pn[2:])) - - -def test_authentication_and_collect(cert_and_key, ip_address): - """Authenticate call and then collect with the returned orderRef UUID.""" - - c = bankid.BankIDJSONClient6(certificates=cert_and_key, test_server=True) - assert "appapi2.test.bankid.com.pem" in c.verify_cert - out = c.authenticate(ip_address, _get_random_personal_number()) - assert isinstance(out, dict) - # UUID.__init__ performs the UUID compliance assertion. - order_ref = uuid.UUID(out.get("orderRef"), version=4) - collect_status = c.collect(out.get("orderRef")) - assert collect_status.get("status") == "pending" - assert collect_status.get("hintCode") in ("outstandingTransaction", "noClient") - - -def test_sign_and_collect(cert_and_key, ip_address): - """Sign call and then collect with the returned orderRef UUID.""" - - c = bankid.BankIDJSONClient6(certificates=cert_and_key, test_server=True) - out = c.sign( - ip_address, - "The data to be signed", - user_non_visible_data="Non visible data", - ) - assert isinstance(out, dict) - # UUID.__init__ performs the UUID compliance assertion. - order_ref = uuid.UUID(out.get("orderRef"), version=4) - collect_status = c.collect(out.get("orderRef")) - assert collect_status.get("status") == "pending" - assert collect_status.get("hintCode") in ("outstandingTransaction", "noClient") - - -def test_invalid_orderref_raises_error(cert_and_key): - c = bankid.BankIDJSONClient6(certificates=cert_and_key, test_server=True) - with pytest.raises(bankid.exceptions.InvalidParametersError): - collect_status = c.collect("invalid-uuid") - - -def test_already_in_progress_raises_error(cert_and_key, ip_address): - c = bankid.BankIDJSONClient6(certificates=cert_and_key, test_server=True) - pn = _get_random_personal_number() - out = c.authenticate(ip_address, requirement={"personalNumber": pn}) - with pytest.raises(bankid.exceptions.AlreadyInProgressError): - out2 = c.authenticate(ip_address, requirement={"personalNumber": pn}) - - -def test_already_in_progress_raises_error_2(cert_and_key, ip_address): - c = bankid.BankIDJSONClient6(certificates=cert_and_key, test_server=True) - pn = _get_random_personal_number() - out = c.sign(ip_address, "Text to sign", requirement={"personalNumber": pn}) - with pytest.raises(bankid.exceptions.AlreadyInProgressError): - out2 = c.sign(ip_address, "Text to sign", requirement={"personalNumber": pn}) - - -def test_authentication_and_cancel(cert_and_key, ip_address): - """Authenticate call and then cancel it""" - - c = bankid.BankIDJSONClient6(certificates=cert_and_key, test_server=True) - out = c.authenticate(ip_address, requirement={"personalNumber": _get_random_personal_number()} ) - assert isinstance(out, dict) - # UUID.__init__ performs the UUID compliance assertion. - order_ref = uuid.UUID(out.get("orderRef"), version=4) - collect_status = c.collect(out.get("orderRef")) - assert collect_status.get("status") == "pending" - assert collect_status.get("hintCode") in ("outstandingTransaction", "noClient") - success = c.cancel(str(order_ref)) - assert success - with pytest.raises(bankid.exceptions.InvalidParametersError): - collect_status = c.collect(out.get("orderRef")) - - -def test_cancel_with_invalid_uuid(cert_and_key): - c = bankid.BankIDJSONClient6(certificates=cert_and_key, test_server=True) - invalid_order_ref = uuid.uuid4() - with pytest.raises(bankid.exceptions.InvalidParametersError): - cancel_status = c.cancel(str(invalid_order_ref)) - - -@pytest.mark.parametrize( - "test_server, endpoint", - [(False, "appapi2.bankid.com"), (True, "appapi2.test.bankid.com")], -) -def test_correct_prod_server_urls(cert_and_key, test_server, endpoint): - c = bankid.BankIDJSONClient6(certificates=cert_and_key, test_server=test_server) - assert c.api_url == "https://{0}/rp/v6.0/".format(endpoint) - assert "{0}.pem".format(endpoint) in c.verify_cert diff --git a/tests/test_jsonclient_sync.py b/tests/test_jsonclient_sync.py deleted file mode 100644 index 3a80552..0000000 --- a/tests/test_jsonclient_sync.py +++ /dev/null @@ -1,79 +0,0 @@ -import uuid - -import pytest - -import bankid - - -def test_authentication_and_collect(cert_and_key, ip_address, random_personal_number): - """Authenticate call and then collect with the returned orderRef UUID.""" - c = bankid.BankIDJSONClient(certificates=cert_and_key, test_server=True) - assert "appapi2.test.bankid.com.pem" in c.verify_cert - out = c.authenticate(ip_address, random_personal_number) - assert isinstance(out, dict) - # UUID.__init__ performs the UUID compliance assertion. - uuid.UUID(out.get("orderRef"), version=4) - collect_status = c.collect(out.get("orderRef")) - assert collect_status.get("status") == "pending" - assert collect_status.get("hintCode") in ("outstandingTransaction", "noClient") - - -def test_sign_and_collect(cert_and_key, ip_address, random_personal_number): - """Sign call and then collect with the returned orderRef UUID.""" - - c = bankid.BankIDJSONClient(certificates=cert_and_key, test_server=True) - out = c.sign( - ip_address, - "The data to be signed", - personal_number=random_personal_number, - user_non_visible_data="Non visible data", - ) - assert isinstance(out, dict) - # UUID.__init__ performs the UUID compliance assertion. - uuid.UUID(out.get("orderRef"), version=4) - collect_status = c.collect(out.get("orderRef")) - assert collect_status.get("status") == "pending" - assert collect_status.get("hintCode") in ("outstandingTransaction", "noClient") - - -def test_invalid_orderref_raises_error(cert_and_key): - c = bankid.BankIDJSONClient(certificates=cert_and_key, test_server=True) - with pytest.raises(bankid.exceptions.InvalidParametersError): - c.collect("invalid-uuid") - - -def test_already_in_progress_raises_error(cert_and_key, ip_address, random_personal_number): - c = bankid.BankIDJSONClient(certificates=cert_and_key, test_server=True) - c.authenticate(ip_address, random_personal_number) - with pytest.raises(bankid.exceptions.AlreadyInProgressError): - c.authenticate(ip_address, random_personal_number) - - -def test_already_in_progress_raises_error_2(cert_and_key, ip_address, random_personal_number): - c = bankid.BankIDJSONClient(certificates=cert_and_key, test_server=True) - c.sign(ip_address, "Text to sign", random_personal_number) - with pytest.raises(bankid.exceptions.AlreadyInProgressError): - c.sign(ip_address, "Text to sign", random_personal_number) - - -def test_authentication_and_cancel(cert_and_key, ip_address, random_personal_number): - """Authenticate call and then cancel it""" - c = bankid.BankIDJSONClient(certificates=cert_and_key, test_server=True) - out = c.authenticate(ip_address, random_personal_number) - assert isinstance(out, dict) - # UUID.__init__ performs the UUID compliance assertion. - order_ref = uuid.UUID(out.get("orderRef"), version=4) - collect_status = c.collect(out.get("orderRef")) - assert collect_status.get("status") == "pending" - assert collect_status.get("hintCode") in ("outstandingTransaction", "noClient") - success = c.cancel(str(order_ref)) - assert success - with pytest.raises(bankid.exceptions.InvalidParametersError): - collect_status = c.collect(out.get("orderRef")) - - -def test_cancel_with_invalid_uuid(cert_and_key): - c = bankid.BankIDJSONClient(certificates=cert_and_key, test_server=True) - invalid_order_ref = uuid.uuid4() - with pytest.raises(bankid.exceptions.InvalidParametersError): - c.cancel(str(invalid_order_ref)) diff --git a/tests/test_syncclient.py b/tests/test_syncclient.py new file mode 100644 index 0000000..13582fb --- /dev/null +++ b/tests/test_syncclient.py @@ -0,0 +1,112 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +:mod:`test_syncclient` +====================== + +.. module:: test_syncclient + :platform: Unix, Windows + :synopsis: + +.. moduleauthor:: mxamin + +Created on 2024-01-18 + +""" +import uuid + +import pytest + +try: + from unittest import mock +except: + import mock + +from bankid import BankIdClient, exceptions + + +def test_authentication_and_collect(cert_and_key, ip_address, random_personal_number): + """Authenticate call and then collect with the returned orderRef UUID.""" + + c = BankIdClient(certificates=cert_and_key, test_server=True) + assert "appapi2.test.bankid.com.pem" in c.verify_cert + out = c.authenticate(ip_address, random_personal_number) + assert isinstance(out, dict) + # UUID.__init__ performs the UUID compliance assertion. + order_ref = uuid.UUID(out.get("orderRef"), version=4) + collect_status = c.collect(out.get("orderRef")) + assert collect_status.get("status") == "pending" + assert collect_status.get("hintCode") in ("outstandingTransaction", "noClient") + + +def test_sign_and_collect(cert_and_key, ip_address): + """Sign call and then collect with the returned orderRef UUID.""" + + c = BankIdClient(certificates=cert_and_key, test_server=True) + out = c.sign( + ip_address, + user_visible_data="The data to be signed", + user_non_visible_data="Non visible data", + ) + assert isinstance(out, dict) + # UUID.__init__ performs the UUID compliance assertion. + order_ref = uuid.UUID(out.get("orderRef"), version=4) + collect_status = c.collect(out.get("orderRef")) + assert collect_status.get("status") == "pending" + assert collect_status.get("hintCode") in ("outstandingTransaction", "noClient") + + +def test_invalid_orderref_raises_error(cert_and_key): + c = BankIdClient(certificates=cert_and_key, test_server=True) + with pytest.raises(exceptions.InvalidParametersError): + collect_status = c.collect("invalid-uuid") + + +def test_already_in_progress_raises_error(cert_and_key, ip_address, random_personal_number): + c = BankIdClient(certificates=cert_and_key, test_server=True) + out = c.authenticate(ip_address, requirement={"personalNumber": random_personal_number}) + with pytest.raises(exceptions.AlreadyInProgressError): + out2 = c.authenticate(ip_address, requirement={"personalNumber": random_personal_number}) + + +def test_already_in_progress_raises_error_2(cert_and_key, ip_address, random_personal_number): + c = BankIdClient(certificates=cert_and_key, test_server=True) + out = c.sign(ip_address, requirement={"personalNumber": random_personal_number}, user_visible_data="Text to sign") + with pytest.raises(exceptions.AlreadyInProgressError): + out2 = c.sign( + ip_address, requirement={"personalNumber": random_personal_number}, user_visible_data="Text to sign" + ) + + +def test_authentication_and_cancel(cert_and_key, ip_address, random_personal_number): + """Authenticate call and then cancel it""" + + c = BankIdClient(certificates=cert_and_key, test_server=True) + out = c.authenticate(ip_address, requirement={"personalNumber": random_personal_number}) + assert isinstance(out, dict) + # UUID.__init__ performs the UUID compliance assertion. + order_ref = uuid.UUID(out.get("orderRef"), version=4) + collect_status = c.collect(out.get("orderRef")) + assert collect_status.get("status") == "pending" + assert collect_status.get("hintCode") in ("outstandingTransaction", "noClient") + success = c.cancel(str(order_ref)) + assert success + with pytest.raises(exceptions.InvalidParametersError): + collect_status = c.collect(out.get("orderRef")) + + +def test_cancel_with_invalid_uuid(cert_and_key): + c = BankIdClient(certificates=cert_and_key, test_server=True) + invalid_order_ref = uuid.uuid4() + with pytest.raises(exceptions.InvalidParametersError): + cancel_status = c.cancel(str(invalid_order_ref)) + + +@pytest.mark.parametrize( + "test_server, endpoint", + [(False, "appapi2.bankid.com"), (True, "appapi2.test.bankid.com")], +) +def test_correct_prod_server_urls(cert_and_key, test_server, endpoint): + c = BankIdClient(certificates=cert_and_key, test_server=test_server) + assert c.api_url == "https://{0}/rp/v6.0/".format(endpoint) + assert "{0}.pem".format(endpoint) in c.verify_cert