Skip to content

Commit

Permalink
Add options for using custom TLS CA bundles (#112)
Browse files Browse the repository at this point in the history
  • Loading branch information
awoimbee authored Nov 28, 2024
1 parent bd2b409 commit 74da848
Show file tree
Hide file tree
Showing 7 changed files with 219 additions and 53 deletions.
14 changes: 1 addition & 13 deletions doc/source/user_guide/config_file.rst
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ Content

You write the configuration file in `TOML <https://toml.io/>`_.
From this file, you can pass parameters for configuring
the :class:`~ansys.simai.core.client.SimAIClient` instance.
the :class:`~ansys.simai.core.client.SimAIClient` instance (see :ref:`configuration`).


Example
Expand All @@ -69,18 +69,6 @@ Example
totp_enabled = true
Proxy
"""""

If your network is situated behind a proxy, you must add its address
in a ``https_proxy`` key in the ``[default]`` block:

.. code-block:: TOML
[default]
organization = "company"
https_proxy = "http://company_proxy_host:3128" # replacing host and port by the real value
Profiles
--------

Expand Down
20 changes: 10 additions & 10 deletions doc/source/user_guide/proxy.rst
Original file line number Diff line number Diff line change
Expand Up @@ -32,17 +32,17 @@ Because your web browser uses a special
`proxy auto-configuration <https://en.wikipedia.org/wiki/Proxy_auto-config>`_ file, the
proxy is not trusted by your computer.

To fix the issue:
There are multiple ways to fix this issue:

1. Extract the certificates used by your company-configured browser on ``https://simai.ansys.com``.
2. Set the ``REQUESTS_CA_BUNDLE`` environment variable:
1. Try ``tls_ca_bundle="system"`` (requires ``python>=3.10``, see :ref:`configuration`).
2. Extract the required CA certificate:

.. code:: python
a. Extract the certificates used by your company-configured browser on ``https://simai.ansys.com``.
b. Set ``tls_ca_bundle`` (or the ``REQUESTS_CA_BUNDLE`` environment variable):

import os
from pathlib import Path
.. code-block:: TOML
os.environ["REQUESTS_CA_BUNDLE"] = Path(
"~/Downloads/ansys-simai-chain.pem"
).expanduser()
client = ansys.simai.core.from_config()
[default]
organization = "company"
tls_ca_bundle = "/home/username/Documents/my_company_proxy_ca_bundle.pem"
3. As a temporary last resort, one can use ``tls_ca_bundle="unsecure-none"`` (contact your IT department).
14 changes: 13 additions & 1 deletion pdm.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 2 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,8 @@ dependencies = [
"wakepy>=0.8.0",
"tqdm>=4.66.1",
"filelock>=3.10.7",
"typing_extensions>=4.12.0"
"typing_extensions>=4.12.0",
"truststore>=0.10.0; python_version >= '3.10'",
]

[tool.flit.module]
Expand Down
22 changes: 21 additions & 1 deletion src/ansys/simai/core/api/mixin.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,19 +22,22 @@

import logging
import os
import ssl
import sys
from io import BytesIO
from pathlib import Path
from typing import Any, BinaryIO, Dict, List, Optional, Union
from urllib.parse import urljoin
from urllib.request import getproxies

import requests
import requests.adapters
from requests.adapters import HTTPAdapter, Retry
from requests_toolbelt import MultipartEncoder, MultipartEncoderMonitor

from ansys.simai.core import __version__
from ansys.simai.core.data.types import APIResponse, File, MonitorCallback
from ansys.simai.core.errors import ConnectionError
from ansys.simai.core.errors import ConfigurationError, ConnectionError
from ansys.simai.core.utils.auth import Authenticator
from ansys.simai.core.utils.configuration import ClientConfig
from ansys.simai.core.utils.files import file_path_to_obj_file
Expand All @@ -43,11 +46,28 @@
logger = logging.getLogger(__name__)


class TruststoreAdapter(HTTPAdapter):
def init_poolmanager(self, *a, **kw):
if sys.version_info < (3, 10):
raise ConfigurationError("The system CA store can only be used with python >= 3.10")

import truststore

ctx = truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
return super().init_poolmanager(*a, **kw, ssl_context=ctx)


class ApiClientMixin:
"""Provides the core that all mixins and the API client are built on."""

def __init__(self, *args, config: ClientConfig): # noqa: D107
self._session = requests.Session()
if config.tls_ca_bundle == "system":
self._session.mount("https://", TruststoreAdapter())
elif config.tls_ca_bundle == "unsecure-none":
self._session.verify = False
elif isinstance(config.tls_ca_bundle, os.PathLike):
self._session.verify = str(config.tls_ca_bundle)

retries = Retry(total=5, backoff_factor=1, status_forcelist=[502, 503, 504])
self._session.mount("http", HTTPAdapter(max_retries=retries))
Expand Down
55 changes: 28 additions & 27 deletions src/ansys/simai/core/utils/configuration.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,8 @@

import hashlib
import logging
from typing import Optional
from os import PathLike
from typing import Literal, Optional, Union
from urllib.parse import urlparse, urlunparse

from pydantic import (
Expand Down Expand Up @@ -79,36 +80,36 @@ def prompt(cls, values, info):


class ClientConfig(BaseModel, extra="allow"):
interactive: Optional[bool] = Field(
default=True, description="If True, it enables interaction with the terminal."
)
url: HttpUrl = Field(
default=HttpUrl("https://api.simai.ansys.com/v2/"),
description="URL to the SimAI API.",
)
organization: str = Field(
default=None,
validate_default=True,
description="Name of the organization(/company) that the user belongs to.",
)
interactive: Optional[bool] = True
"If True, it enables interaction with the terminal."
url: HttpUrl = HttpUrl("https://api.simai.ansys.com/v2/")
"URL to the SimAI API."
organization: str = Field(None, validate_default=True)
"Name of the organization(/company) that the user belongs to."
workspace: Optional[str] = None
"Name of the workspace to use by default."
project: Optional[str] = None
"Name of the project to use by default."
credentials: Optional[Credentials] = Field(
default=None,
validate_default=True,
description="Authenticate via username/password instead of the device authorization code.",
)
workspace: Optional[str] = Field(
default=None, description="Name of the workspace to use by default."
)
project: Optional[str] = Field(
default=None, description="Name of the project to use by default."
)
https_proxy: Optional[AnyHttpUrl] = Field(
default=None, description="URL of the HTTPS proxy to use."
)
skip_version_check: bool = Field(default=False, description="Skip checking for updates.")
no_sse_connection: bool = Field(
default=False, description="Don't receive live updates from the SimAI API."
)
"Authenticate via username/password instead of the device authorization code."
skip_version_check: bool = False
"Skip checking for updates."
no_sse_connection: bool = False
"Don't receive live updates from the SimAI API."
https_proxy: Optional[AnyHttpUrl] = None
"URL of the HTTPS proxy to use."
tls_ca_bundle: Union[Literal["system", "unsecure-none"], PathLike, None] = None
"""
Custom TLS CA certificate configuration. Possible values:
* ``None``: use secure defaults
* ``"system"``: uses system CA certificates (python >= 3.10)
* A ``PathLike`` object: use a custom CA
* ``"unsecure-none"``: no TLS certificate validation
"""

@field_validator("url", mode="before")
def clean_url(cls, url):
Expand Down
144 changes: 144 additions & 0 deletions tests/test_client_config_tls_ca_bundle.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,144 @@
# Copyright (C) 2024 ANSYS, Inc. and/or its affiliates.
# SPDX-License-Identifier: MIT
#
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import contextlib
import os
import ssl
import subprocess
import sys
import threading
from http.server import HTTPServer, SimpleHTTPRequestHandler
from unittest.mock import patch

import pytest

import ansys.simai.core.errors as err
from ansys.simai.core import SimAIClient


#
# SETUP
#
@pytest.fixture(scope="module")
def tls_root_certificate(tmp_path_factory):
tmp_path = tmp_path_factory.mktemp("tls_root_certificate")
# Create CA key and certificate
subprocess.run(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", tmp_path / "ca_key.pem", "-out", tmp_path / "test_ca.pem", "-days", "3650", "-subj", "/CN=Test CA"], check=False) # fmt: skip # noqa: S607, S603
# Create server key and certificate
subprocess.run(["openssl", "req", "-newkey", "rsa:2048", "-nodes", "-keyout", tmp_path / "server_key.pem", "-out", tmp_path / "server_csr.pem", "-subj", "/CN=localhost"], check=False) # fmt: skip # noqa: S607, S603
with open(tmp_path / "openssl.cnf", "w") as f:
# Create a config file for OpenSSL to include the SAN
f.write("[SAN]\nsubjectAltName = DNS:localhost\n")
subprocess.run(["openssl", "x509", "-req", "-in", tmp_path / "server_csr.pem", "-CA", tmp_path / "test_ca.pem", "-CAkey", tmp_path / "ca_key.pem", "-CAcreateserial", "-out", tmp_path / "server_cert.pem", "-days", "3650", "-extensions", "SAN", "-extfile", tmp_path / "openssl.cnf"], check=False) # fmt: skip # noqa: S607, S603

return {
"ca": tmp_path / "test_ca.pem",
"server_key": tmp_path / "server_key.pem",
"server_cert": tmp_path / "server_cert.pem",
}


@pytest.fixture(scope="module")
def https_server(tls_root_certificate):
class SimpleHTTPSHandler(SimpleHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.send_header("Content-type", "text/plain")
self.end_headers()
self.wfile.write(b"Hello, Secure World!")

ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_context.load_cert_chain(
certfile=tls_root_certificate["server_cert"], keyfile=tls_root_certificate["server_key"]
)
httpd = HTTPServer(("localhost", 48219), SimpleHTTPSHandler)
httpd.socket = ssl_context.wrap_socket(
httpd.socket,
server_side=True,
)
server_thread = threading.Thread(target=httpd.serve_forever, daemon=True)
server_thread.start()
yield "https://localhost:48219"
httpd.shutdown()


BASE_CLT_ARGS = {
"url": "https://test.example.com",
"organization": "dummy",
"_disable_authentication": True,
"no_sse_connection": True,
"skip_version_check": True,
}


#
# TESTS
#
def test_client_without_config_tls_ca_bundle(tls_root_certificate, https_server):
# Default config, using certifi's certificate store
clt = SimAIClient(**BASE_CLT_ARGS)
# By default, the self-signed test CA is not accepted
with pytest.raises(err.ConnectionError):
clt._api._get(https_server)
# One can use REQUESTS_CA_BUNDLE
with patch.dict(os.environ, {"REQUESTS_CA_BUNDLE": str(tls_root_certificate["ca"])}):
clt._api._get(https_server, return_json=False)


@pytest.mark.skipif(
sys.version_info < (3, 11),
reason='"system" requires Python >= 3.10, "patch" is broken in python 3.10',
)
def test_client_config_tls_ca_bundle_system(tls_root_certificate, https_server):
clt = SimAIClient(**BASE_CLT_ARGS, tls_ca_bundle="system")
# The system CA rejects the test CA by default
with pytest.raises(err.ConnectionError):
clt._api._get(https_server)

# If the host system trusts the test CA, pysimai trusts it !
@contextlib.contextmanager
def load_test_ca_as_truststore_system_ca(ctx: "ssl.SSLContext"):
ctx.load_verify_locations(cafile=tls_root_certificate["ca"])
yield

with patch(
"truststore._api._configure_context", side_effect=load_test_ca_as_truststore_system_ca
):
clt._api._get(https_server, return_json=False)


@pytest.mark.skipif(
sys.version_info >= (3, 10), reason="The error is only raised for python < 3.10"
)
def test_client_config_tls_ca_bundle_system_on_usupported_python_version():
with pytest.raises(err.ConfigurationError, match="python >= 3.10"):
SimAIClient(**BASE_CLT_ARGS, tls_ca_bundle="system")


def test_client_config_tls_ca_bundle_unsecure_none(https_server):
clt = SimAIClient(**BASE_CLT_ARGS, tls_ca_bundle="unsecure-none")
clt._api._get(https_server, return_json=False)


def test_client_config_tls_ca_bundle_path(tls_root_certificate, https_server):
clt = SimAIClient(**BASE_CLT_ARGS, tls_ca_bundle=tls_root_certificate["ca"])
clt._api._get(https_server, return_json=False)

0 comments on commit 74da848

Please sign in to comment.