Remove pyoidc and pysaml2 from core

setup.py

@@ -17,7 +17,7 @@
     install_requires=[
         "pyop >= v3.4.0",
         "pysaml2 >= 6.5.1",
-        "pycryptodomex",
+        "cryptojwt >= 1.8.3",
         "requests",
         "PyYAML",
         "gunicorn",

src/satosa/backends/apple.py

@@ -258,7 +258,7 @@ def _translate_response(self, response, issuer):
         :type subject_type: str
         :rtype: InternalData
 
-        :param response: Dictioary with attribute name as key.
+        :param response: Dictionary with attribute name as key.
         :param issuer: The oidc op that gave the repsonse.
         :param subject_type: public or pairwise according to oidc standard.
         :return: A SATOSA internal response.

src/satosa/backends/base.py

@@ -66,7 +66,7 @@ def register_endpoints(self):
     def get_metadata_desc(self):
         """
         Returns a description of the backend module.
-        This is used when creating SAML metadata for the frontend of the proxy
+
         :rtype: satosa.metadata_creation.description.MetadataDescription
         :return: A description of the backend
         """

src/satosa/backends/idpy_oidc.py

@@ -1,22 +1,21 @@
 """
 OIDC/OAuth2 backend module.
 """
-import datetime
 import logging
+from datetime import datetime
 from urllib.parse import urlparse
 
 from idpyoidc.client.oauth2.stand_alone_client import StandAloneClient
 from idpyoidc.server.user_authn.authn_context import UNSPECIFIED
 
+import satosa.logging_util as lu
 from satosa.backends.base import BackendModule
 from satosa.internal import AuthenticationInformation
 from satosa.internal import InternalData
-import satosa.logging_util as lu
 from ..exception import SATOSAAuthenticationError
 from ..exception import SATOSAError
 from ..response import Redirect
 
-
 UTC = datetime.timezone.utc
 logger = logging.getLogger(__name__)
 
@@ -48,14 +47,7 @@ def __init__(self, auth_callback_func, internal_attributes, config, base_url, na
         super().__init__(auth_callback_func, internal_attributes, base_url, name)
         # self.auth_callback_func = auth_callback_func
         # self.config = config
-        self.client = StandAloneClient(config=config["client"], client_type="oidc")
-        self.client.do_provider_info()
-        self.client.do_client_registration()
-
-        _redirect_uris = self.client.context.claims.get_usage('redirect_uris')
-        if not _redirect_uris:
-            raise SATOSAError("Missing path in redirect uri")
-        self.redirect_path = urlparse(_redirect_uris[0]).path
+        self.client = create_client(config["client"])
 
     def start_auth(self, context, internal_request):
         """
@@ -77,7 +69,11 @@ def register_endpoints(self):
         :return: A list that can be used to map the request to SATOSA to this endpoint.
         """
         url_map = []
-        url_map.append((f"^{self.redirect_path.lstrip('/')}$", self.response_endpoint))
+        redirect_path = self.client.context.claims.get_usage('redirect_uris')
+        if not redirect_path:
+            raise SATOSAError("Missing path in redirect uri")
+        redirect_path = urlparse(redirect_path[0]).path
+        url_map.append(("^%s$" % redirect_path.lstrip("/"), self.response_endpoint))
         return url_map
 
     def response_endpoint(self, context, *args):
@@ -123,16 +119,7 @@ def _translate_response(self, response, issuer):
         :param subject_type: public or pairwise according to oidc standard.
         :return: A SATOSA internal response.
         """
-        timestamp_epoch = (
-            response.get("auth_time")
-            or response.get("iat")
-            or int(datetime.datetime.now(UTC).timestamp())
-        )
-        timestamp_dt = datetime.datetime.fromtimestamp(timestamp_epoch, UTC)
-        timestamp_iso = timestamp_dt.isoformat().replace("+00:00", "Z")
-        auth_class_ref = response.get("acr") or response.get("amr") or UNSPECIFIED
-        auth_info = AuthenticationInformation(auth_class_ref, timestamp_iso, issuer)
-
+        auth_info = AuthenticationInformation(UNSPECIFIED, str(datetime.now()), issuer)
         internal_resp = InternalData(auth_info=auth_info)
         internal_resp.attributes = self.converter.to_internal("openid", response)
         internal_resp.subject_id = response["sub"]
@@ -154,3 +141,10 @@ def _check_error_response(self, response, context):
             logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
             logger.debug(logline)
             raise SATOSAAuthenticationError(context.state, "Access denied")
+
+def create_client(config: dict):
+    _client_type = config.get('client_type') or "oidc"
+    _client = StandAloneClient(config=config, client_type=_client_type)
+    _client.do_provider_info()
+    _client.do_client_registration()
+    return _client

src/satosa/base.py

@@ -5,7 +5,7 @@
 import logging
 import uuid
 
-from saml2.s_utils import UnknownSystemEntity
+# from saml2.s_utils import UnknownSystemEntity
 
 from satosa import util
 from satosa.response import BadRequest
@@ -309,48 +309,11 @@ def run(self, context):
                 redirect_url = f"{generic_error_url}?errorid={error_id}"
                 return Redirect(generic_error_url)
             raise
-        except SATOSANoBoundEndpointError as e:
-            error_id = uuid.uuid4().urn
-            msg = {
-                "message": "URL-path is not bound to any endpoint function",
-                "error": str(e),
-                "error_id": error_id,
-            }
-            logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
-            logger.error(logline)
-            generic_error_url = self.config.get("ERROR_URL")
-            if generic_error_url:
-                redirect_url = f"{generic_error_url}?errorid={error_id}"
-                return Redirect(generic_error_url)
-            return NotFound("The Service or Identity Provider you requested could not be found.")
-        except SATOSAError as e:
-            error_id = uuid.uuid4().urn
-            msg = {
-                "message": "Uncaught SATOSA error",
-                "error": str(e),
-                "error_id": error_id,
-            }
-            logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
-            logger.error(logline)
-            generic_error_url = self.config.get("ERROR_URL")
-            if generic_error_url:
-                redirect_url = f"{generic_error_url}?errorid={error_id}"
-                return Redirect(generic_error_url)
-            raise
-        except UnknownSystemEntity as e:
-            error_id = uuid.uuid4().urn
-            msg = {
-                "message": "Configuration error: unknown system entity",
-                "error": str(e),
-                "error_id": error_id,
-            }
-            logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
-            logger.error(logline)
-            generic_error_url = self.config.get("ERROR_URL")
-            if generic_error_url:
-                redirect_url = f"{generic_error_url}?errorid={error_id}"
-                return Redirect(generic_error_url)
-            raise
+        # except UnknownSystemEntity as err:
+        #     msg = "configuration error: unknown system entity " + str(err)
+        #     logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
+        #     logger.error(logline, exc_info=False)
+        #     raise
         except Exception as e:
             error_id = uuid.uuid4().urn
             msg = {

src/satosa/cert_util.py

@@ -0,0 +1,75 @@
+import datetime
+
+from cryptography import x509
+from cryptography.hazmat._oid import NameOID
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptojwt.jwk.rsa import import_private_rsa_key_from_file
+from cryptojwt.jwk.rsa import RSAKey
+
+
+def create_certificate(cert_info):
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+
+    subject = issuer = x509.Name([
+        x509.NameAttribute(NameOID.COUNTRY_NAME, cert_info['cn']),
+        x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, cert_info['state']),
+        x509.NameAttribute(NameOID.LOCALITY_NAME, cert_info['state']),
+        x509.NameAttribute(NameOID.ORGANIZATION_NAME, cert_info['organization']),
+        x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, cert_info['organization_unit']),
+    ])
+    item = x509.CertificateBuilder().subject_name(
+        subject
+    ).issuer_name(
+        issuer
+    ).public_key(
+        key.public_key()
+    ).serial_number(
+        x509.random_serial_number()
+    ).not_valid_before(
+        datetime.datetime.utcnow()
+    ).not_valid_after(
+        datetime.datetime.utcnow() + datetime.timedelta(days=10)
+    )
+
+    if 'dns_name' in cert_info:
+        item.add_extension(
+            x509.SubjectAlternativeName([x509.DNSName(cert_info['dns_name'])]), critical=False
+        )
+
+    cert = item.sign(key, hashes.SHA256())
+    cert_str = cert.public_bytes(serialization.Encoding.PEM)
+
+    key_str = key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.TraditionalOpenSSL,
+        encryption_algorithm=serialization.NoEncryption(),
+    )
+    return cert_str, key_str
+
+
+def generate_cert():
+    cert_info = {
+        "cn": "SE",
+        "country_code": "se",
+        "state": "ac",
+        "city": "Umea",
+        "organization": "ITS",
+        "organization_unit": "DIRG"
+    }
+    cert_str, key_str = create_certificate(cert_info)
+    return cert_str, key_str
+
+
+def write_cert(cert_path, key_path):
+    cert, key = generate_cert()
+    with open(cert_path, "wb") as cert_file:
+        cert_file.write(cert)
+    with open(key_path, "wb") as key_file:
+        key_file.write(key)
+
+def rsa_key_from_pem(file_name, **kwargs):
+    _key = RSAKey(**kwargs)
+    _key.load_key(import_private_rsa_key_from_file(file_name))
+    return _key

src/satosa/micro_services/account_linking.py

@@ -5,10 +5,11 @@
 import logging
 
 import requests
-from jwkest.jwk import rsa_load, RSAKey
-from jwkest.jws import JWS
+from cryptojwt import JWS
+from cryptojwt.jwk.rsa import import_private_rsa_key_from_file
 
 from satosa.internal import InternalData
+from ..cert_util import rsa_key_from_pem
 from ..exception import SATOSAAuthenticationError
 from ..micro_services.base import ResponseMicroService
 from ..response import Redirect
@@ -30,7 +31,8 @@ def __init__(self, config, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.api_url = config["api_url"]
         self.redirect_url = config["redirect_url"]
-        self.signing_key = RSAKey(key=rsa_load(config["sign_key"]), use="sig", alg="RS256")
+        self.signing_key = rsa_key_from_pem(config["sign_key"])
+        self.signing_key.alg = config.get('signing_alg', "RS256")
         self.endpoint = "/handle_account_linking"
         self.id_to_attr = config.get("id_to_attr", None)
         logger.info("Account linking is active")

src/satosa/micro_services/consent.py

@@ -7,12 +7,16 @@
 from base64 import urlsafe_b64encode
 
 import requests
-from jwkest.jwk import RSAKey
-from jwkest.jwk import rsa_load
-from jwkest.jws import JWS
+from cryptojwt import JWS
+from cryptojwt.jwk.rsa import import_private_rsa_key_from_file
+from cryptojwt.jwk.rsa import RSAKey
+# from jwkest.jwk import RSAKey
+# from jwkest.jwk import rsa_load
+# from jwkest.jws import JWS
 from requests.exceptions import ConnectionError
 
 import satosa.logging_util as lu
+from satosa.cert_util import rsa_key_from_pem
 from satosa.internal import InternalData
 from satosa.micro_services.base import ResponseMicroService
 from satosa.response import Redirect
@@ -41,7 +45,7 @@ def __init__(self, config, internal_attributes, *args, **kwargs):
         if "user_id_to_attr" in internal_attributes:
             self.locked_attr = internal_attributes["user_id_to_attr"]
 
-        self.signing_key = RSAKey(key=rsa_load(config["sign_key"]), use="sig", alg="RS256")
+        self.signing_key = rsa_key_from_pem(config["sign_key"], use="sig", alg="RS256")
         self.endpoint = "/handle_consent"
         logger.info("Consent flow is active")