Skip to content

Commit

Permalink
Feat(anta.cli): Make CV certs verified by default
Browse files Browse the repository at this point in the history
  • Loading branch information
gmuloc committed Jun 7, 2024
1 parent 822e5e0 commit 5d50a3b
Show file tree
Hide file tree
Showing 4 changed files with 44 additions and 13 deletions.
11 changes: 9 additions & 2 deletions anta/cli/get/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,14 +36,21 @@
@click.option("--username", "-u", help="CloudVision username", type=str, required=True)
@click.option("--password", "-p", help="CloudVision password", type=str, required=True)
@click.option("--container", "-c", help="CloudVision container where devices are configured", type=str)
def from_cvp(ctx: click.Context, output: Path, host: str, username: str, password: str, container: str | None) -> None:
@click.option(
"--ignore-cert",
help="By default connection to CV will use HTTPS certificate, set this flag to disable it",
show_envvar=True,
is_flag=True,
default=False,
)
def from_cvp(ctx: click.Context, output: Path, host: str, username: str, password: str, container: str | None, *, ignore_cert: bool) -> None:
# pylint: disable=too-many-arguments
"""Build ANTA inventory from Cloudvision.
TODO - handle get_inventory and get_devices_in_container failure
"""
logger.info("Getting authentication token for user '%s' from CloudVision instance '%s'", username, host)
token = get_cv_token(cvp_ip=host, cvp_username=username, cvp_password=password)
token = get_cv_token(cvp_ip=host, cvp_username=username, cvp_password=password, verify_cert=not ignore_cert)

clnt = CvpClient()
try:
Expand Down
21 changes: 17 additions & 4 deletions anta/cli/get/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -77,16 +77,29 @@ def wrapper(
return wrapper


def get_cv_token(cvp_ip: str, cvp_username: str, cvp_password: str) -> str:
"""Generate AUTH token from CVP using password."""
# TODO: need to handle requests error
def get_cv_token(cvp_ip: str, cvp_username: str, cvp_password: str, *, verify_cert: bool) -> str:
"""Generate AUTH token from CVP using password.
TODO: need to handle requests error
Args:
----
cvp_ip: IP address of CloudVision.
cvp_username: Username to connect to CloudVision.
cvp_password: Password to connect to CloudVision.
verify_cert: Enable or disable certificate verification when connecting to CloudVision.
Returns
-------
token(str): The token to use in further API calls to CloudVision.
"""
# use CVP REST API to generate a token
url = f"https://{cvp_ip}/cvpservice/login/authenticate.do"
payload = json.dumps({"userId": cvp_username, "password": cvp_password})
headers = {"Content-Type": "application/json", "Accept": "application/json"}

response = requests.request("POST", url, headers=headers, data=payload, verify=False, timeout=10)
response = requests.request("POST", url, headers=headers, data=payload, verify=verify_cert, timeout=10)
return response.json()["sessionId"]


Expand Down
12 changes: 8 additions & 4 deletions tests/units/cli/get/test_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,17 +24,19 @@


@pytest.mark.parametrize(
("cvp_container", "cvp_connect_failure"),
("cvp_container", "verify_cert", "cvp_connect_failure"),
[
pytest.param(None, False, id="all devices"),
pytest.param("custom_container", False, id="custom container"),
pytest.param(None, True, id="cvp connect failure"),
pytest.param(None, True, False, id="all devices - verify cert"),
pytest.param(None, False, False, id="all devices - do not verify cert"),
pytest.param("custom_container", False, False, id="custom container"),
pytest.param(None, False, True, id="cvp connect failure"),
],
)
def test_from_cvp(
tmp_path: Path,
click_runner: CliRunner,
cvp_container: str | None,
verify_cert: bool,
cvp_connect_failure: bool,
) -> None:
"""Test `anta get from-cvp`.
Expand All @@ -57,6 +59,8 @@ def test_from_cvp(

if cvp_container is not None:
cli_args.extend(["--container", cvp_container])
if not verify_cert:
cli_args.extend(["--ignore-cert"])

def mock_cvp_connect(_self: CvpClient, *_args: str, **_kwargs: str) -> None:
if cvp_connect_failure:
Expand Down
13 changes: 10 additions & 3 deletions tests/units/cli/get/test_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,14 @@
DATA_DIR: Path = Path(__file__).parents[3].resolve() / "data"


def test_get_cv_token() -> None:
@pytest.mark.parametrize(
"verify_cert",
[
pytest.param(True, id="Verify cert enabled"),
pytest.param(False, id="Verify cert disabled"),
],
)
def test_get_cv_token(verify_cert: bool) -> None:
"""Test anta.get.utils.get_cv_token."""
ip_addr = "42.42.42.42"
username = "ant"
Expand All @@ -29,13 +36,13 @@ def test_get_cv_token() -> None:
mocked_ret = MagicMock(autospec=requests.Response)
mocked_ret.json.return_value = {"sessionId": "simple"}
patched_request.return_value = mocked_ret
res = get_cv_token(ip_addr, username, password)
res = get_cv_token(ip_addr, username, password, verify_cert=verify_cert)
patched_request.assert_called_once_with(
"POST",
"https://42.42.42.42/cvpservice/login/authenticate.do",
headers={"Content-Type": "application/json", "Accept": "application/json"},
data='{"userId": "ant", "password": "formica"}',
verify=False,
verify=verify_cert,
timeout=10,
)
assert res == "simple"
Expand Down

0 comments on commit 5d50a3b

Please sign in to comment.