Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Verify TLS by default to avoid warnings #299

Merged
merged 1 commit into from
Aug 15, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ pylint:

.PHONY: flake8
flake8:
flake8 --max-line-length=$(LINE_MAX) .
flake8 --max-line-length=$(LINE_MAX) $(FILES) tests/*.py
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we should keep nice and tidy model which we HAD , with variables . if FILES variable is not enough better create one more variable ...


.PHONY: test
test:
Expand Down
25 changes: 16 additions & 9 deletions ocw/lib/openqa.py
Original file line number Diff line number Diff line change
@@ -1,29 +1,36 @@
from urllib.parse import urlparse
from cachetools import cached
from requests.exceptions import RequestException
import requests
from requests.exceptions import RequestException, SSLError
import openqa_client.client
from openqa_client.const import JOB_STATE_CANCELLED
from openqa_client.exceptions import OpenQAClientError


# We don't verify TLS server certificates because we
# may encounter self-signed or expired certificates
DEFAULT_VERIFY = False
DEFAULT_TIMEOUT = 3
asmorodskyi marked this conversation as resolved.
Show resolved Hide resolved


def verify_tls(url: str) -> bool:
try:
requests.head(url, timeout=DEFAULT_TIMEOUT, verify=True).raise_for_status()
except (RequestException, SSLError):
return False
return True


@cached(cache={})
def get_url(server):
def get_url(server: str) -> str:
if server:
server = server.rstrip('/').replace("_", ".")
if urlparse(server).scheme != "":
return server
for scheme in ("https", "http"):
try:
url = f"{scheme}://{server}"
got = openqa_client.client.requests.head(
got = requests.head(
url,
timeout=5,
verify=DEFAULT_VERIFY,
timeout=DEFAULT_TIMEOUT,
verify=False,
)
got.raise_for_status()
return url
Expand All @@ -45,7 +52,7 @@ def __new__(cls, **kwargs):
def __init__(self, **kwargs):
kwargs.pop("server")
self.__client = openqa_client.client.OpenQA_Client(server=self.server, **kwargs)
self.__client.session.verify = DEFAULT_VERIFY
self.__client.session.verify = verify_tls(self.server)

def is_cancelled(self, job_id: str) -> bool:
if not job_id.isdigit():
Expand Down
30 changes: 27 additions & 3 deletions tests/test_openqa.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,8 @@
import pytest
from openqa_client.const import JOB_STATE_CANCELLED
from openqa_client.exceptions import OpenQAClientError
from requests.exceptions import RequestException
from ocw.lib.openqa import OpenQA, get_url
from requests.exceptions import RequestException, SSLError
from ocw.lib.openqa import OpenQA, get_url, verify_tls


@pytest.fixture
Expand All @@ -15,7 +15,8 @@ def openqa_client_mock():
def openqa_instance(openqa_client_mock):
with (
patch('openqa_client.client.OpenQA_Client', return_value=openqa_client_mock),
patch('ocw.lib.openqa.get_url', return_value=None)
patch('ocw.lib.openqa.get_url', return_value=None),
patch('ocw.lib.openqa.verify_tls', return_value=None),
):
yield OpenQA(server="myserver")

Expand Down Expand Up @@ -109,3 +110,26 @@ def test_get_url_with_invalid_server():
with pytest.raises(OpenQAClientError):
get_url("invalid-openqa.suse.de")
get_url.cache_clear()


# Create a mock response for requests.head
class MockResponse:
def raise_for_status(self):
pass


def test_verify_tls_valid_url():
with patch("requests.head", return_value=MockResponse()):
assert verify_tls("https://www.example.com")


def test_verify_tls_invalid_url():
with patch("requests.head", side_effect=SSLError()):
result = verify_tls("https://www.invalidurl.com")
assert not result


def test_verify_tls_request_exception():
with patch("requests.head", side_effect=RequestException()):
result = verify_tls("https://www.example.com")
assert not result