test_container_management.py

"""Tests for the Container Management Content

:Requirement: ContainerImageManagement

:CaseAutomation: Automated

:Team: Artemis

:CaseComponent: ContainerImageManagement

"""

from datetime import UTC, datetime
import re

from box import Box
from fauxfactory import gen_string
import pytest
from wait_for import wait_for

from robottelo.config import settings
from robottelo.constants import REPO_TYPE
from robottelo.logging import logger
from robottelo.utils.issue_handlers import is_open


def _repo(sat, product_id, name=None, upstream_name=None, url=None):
    """Creates a Docker-based repository.

    :param product_id: ID of the ``Product``.
    :param str name: Name for the repository. If ``None`` then a random
        value will be generated.
    :param str upstream_name: A valid name of an existing upstream repository.
        If ``None`` then defaults to settings.container.upstream_name constant.
    :param str url: URL of repository. If ``None`` then defaults to
        settings.container.registry_hub constant.
    :return: A ``Repository`` object.
    """
    return sat.cli_factory.make_repository(
        {
            'content-type': REPO_TYPE['docker'],
            'docker-upstream-name': upstream_name or settings.container.upstream_name,
            'name': name or gen_string('alpha', 5),
            'product-id': product_id,
            'url': url or settings.container.registry_hub,
        }
    )


class TestDockerClient:
    """Tests specific to using ``Docker`` as a client to pull Docker images
    from a Satellite 6 instance.

    :CaseImportance: Medium
    """

    def test_positive_pull_image(
        self, request, module_org, module_container_contenthost, target_sat
    ):
        """A Docker-enabled client can use ``docker pull`` to pull a
        Docker image off a Satellite 6 instance.

        :id: 023f0538-2aad-4f87-b8a8-6ccced648366

        :steps:

            1. Publish and promote content view with Docker content
            2. Register Docker-enabled client against Satellite 6.

        :expectedresults: Client can pull Docker images from server and run it.

        :parametrized: yes
        """
        product = target_sat.cli_factory.make_product_wait({'organization-id': module_org.id})
        repo = _repo(target_sat, product['id'])
        target_sat.cli.Repository.synchronize({'id': repo['id']})
        repo = target_sat.cli.Repository.info({'id': repo['id']})
        try:
            result = module_container_contenthost.execute(
                f'docker login -u {settings.server.admin_username}'
                f' -p {settings.server.admin_password} {target_sat.hostname}'
            )
            assert result.status == 0
            request.addfinalizer(
                lambda: module_container_contenthost.execute(f'docker logout {target_sat.hostname}')
            )

            # publishing takes few seconds sometimes
            result, _ = wait_for(
                lambda: module_container_contenthost.execute(f'docker pull {repo["published-at"]}'),
                num_sec=60,
                delay=2,
                fail_condition=lambda out: out.status != 0,
                logger=logger,
            )
            assert result.status == 0
            try:
                result = module_container_contenthost.execute(
                    f'docker run -d {repo["published-at"]}'
                )
                assert result.status == 0
                match = re.match(r'^[0-9a-f]+$', result.stdout)
                if match:
                    container_id = match.group(0)
            finally:
                # Stop and remove the container
                if container_id:
                    module_container_contenthost.execute(f'docker stop {container_id}')
                    module_container_contenthost.execute(f'docker rm {container_id}')
        finally:
            # Remove docker image
            module_container_contenthost.execute(f'docker rmi {repo["published-at"]}')

    @pytest.mark.skip_if_not_set('docker')
    @pytest.mark.e2e
    def test_positive_container_admin_end_to_end_search(
        self, request, module_org, module_container_contenthost, target_sat
    ):
        """Verify that docker command line can be used against
        Satellite server to search for container images stored
        on Satellite instance.

        :id: cefa74e1-e40d-4f47-853b-1268643cea2f

        :steps:

            1. Publish and promote content view with Docker content
            2. Set 'Unauthenticated Pull' option to false
            3. Try to search for docker images on Satellite
            4. Use Docker client to login to Satellite docker hub
            5. Search for docker images
            6. Use Docker client to log out of Satellite docker hub
            7. Try to search for docker images (ensure last search result
               is caused by change of Satellite option and not login/logout)
            8. Set 'Unauthenticated Pull' option to true
            9. Search for docker images

        :expectedresults: Client can search for docker images stored
            on Satellite instance

        :parametrized: yes
        """
        pattern_prefix = gen_string('alpha', 5)
        registry_name_pattern = (
            f'{pattern_prefix}-<%= content_view.label %>/<%= repository.docker_upstream_name %>'
        )

        # Satellite setup: create product and add Docker repository;
        # create content view and add Docker repository;
        # create lifecycle environment and promote content view to it
        lce = target_sat.cli_factory.make_lifecycle_environment({'organization-id': module_org.id})
        product = target_sat.cli_factory.make_product_wait({'organization-id': module_org.id})
        repo = _repo(target_sat, product['id'], upstream_name=settings.container.upstream_name)
        target_sat.cli.Repository.synchronize({'id': repo['id']})
        content_view = target_sat.cli_factory.make_content_view(
            {'composite': False, 'organization-id': module_org.id}
        )
        target_sat.cli.ContentView.add_repository(
            {'id': content_view['id'], 'repository-id': repo['id']}
        )
        target_sat.cli.ContentView.publish({'id': content_view['id']})
        content_view = target_sat.cli.ContentView.info({'id': content_view['id']})
        target_sat.cli.ContentView.version_promote(
            {'id': content_view['versions'][0]['id'], 'to-lifecycle-environment-id': lce['id']}
        )
        target_sat.cli.LifecycleEnvironment.update(
            {
                'registry-name-pattern': registry_name_pattern,
                'registry-unauthenticated-pull': 'false',
                'id': lce['id'],
                'organization-id': module_org.id,
            }
        )
        docker_repo_uri = (
            f'{target_sat.hostname}/{pattern_prefix}-{content_view["label"]}/'
            f'{settings.container.upstream_name}'
        ).lower()

        # 3. Try to search for docker images on Satellite
        remote_search_command = (
            f'docker search {target_sat.hostname}/{settings.container.upstream_name}'
        )
        result = module_container_contenthost.execute(remote_search_command)
        assert result.status == 0
        assert docker_repo_uri not in result.stdout

        # 4. Use Docker client to login to Satellite docker hub
        result = module_container_contenthost.execute(
            f'docker login -u {settings.server.admin_username}'
            f' -p {settings.server.admin_password} {target_sat.hostname}'
        )
        assert result.status == 0
        request.addfinalizer(
            lambda: module_container_contenthost.execute(f'docker logout {target_sat.hostname}')
        )

        # 5. Search for docker images
        result = module_container_contenthost.execute(remote_search_command)
        assert result.status == 0
        assert docker_repo_uri in result.stdout

        # 6. Use Docker client to log out of Satellite docker hub
        result = module_container_contenthost.execute(f'docker logout {target_sat.hostname}')
        assert result.status == 0

        # 7. Try to search for docker images
        result = module_container_contenthost.execute(remote_search_command)
        assert result.status == 0
        assert docker_repo_uri not in result.stdout

        # 8. Set 'Unauthenticated Pull' option to true
        target_sat.cli.LifecycleEnvironment.update(
            {
                'registry-unauthenticated-pull': 'true',
                'id': lce['id'],
                'organization-id': module_org.id,
            }
        )

        # 9. Search for docker images
        result = module_container_contenthost.execute(remote_search_command)
        assert result.status == 0
        assert docker_repo_uri in result.stdout

    @pytest.mark.skip_if_not_set('docker')
    @pytest.mark.e2e
    def test_positive_container_admin_end_to_end_pull(
        self, request, module_org, module_container_contenthost, target_sat
    ):
        """Verify that docker command line can be used against
        Satellite server to pull in container images stored
        on Satellite instance.

        :id: 2a331f88-406b-4a5c-ae70-302a9994077f

        :steps:

            1. Publish and promote content view with Docker content
            2. Set 'Unauthenticated Pull' option to false
            3. Try to pull in docker image from Satellite
            4. Use Docker client to login to Satellite container registry
            5. Pull in docker image
            6. Use Docker client to log out of Satellite container registry
            7. Try to pull in docker image (ensure next pull result
               is caused by change of Satellite option and not login/logout)
            8. Set 'Unauthenticated Pull' option to true
            9. Pull in docker image

        :expectedresults: Client can pull in docker images stored
            on Satellite instance

        :parametrized: yes
        """
        pattern_prefix = gen_string('alpha', 5)
        docker_upstream_name = settings.container.upstream_name
        registry_name_pattern = (
            f'{pattern_prefix}-<%= content_view.label %>/<%= repository.docker_upstream_name %>'
        )

        # Satellite setup: create product and add Docker repository;
        # create content view and add Docker repository;
        # create lifecycle environment and promote content view to it
        lce = target_sat.cli_factory.make_lifecycle_environment({'organization-id': module_org.id})
        product = target_sat.cli_factory.make_product_wait({'organization-id': module_org.id})
        repo = _repo(target_sat, product['id'], upstream_name=docker_upstream_name)
        target_sat.cli.Repository.synchronize({'id': repo['id']})
        content_view = target_sat.cli_factory.make_content_view(
            {'composite': False, 'organization-id': module_org.id}
        )
        target_sat.cli.ContentView.add_repository(
            {'id': content_view['id'], 'repository-id': repo['id']}
        )
        target_sat.cli.ContentView.publish({'id': content_view['id']})
        content_view = target_sat.cli.ContentView.info({'id': content_view['id']})
        target_sat.cli.ContentView.version_promote(
            {'id': content_view['versions'][0]['id'], 'to-lifecycle-environment-id': lce['id']}
        )
        target_sat.cli.LifecycleEnvironment.update(
            {
                'registry-name-pattern': registry_name_pattern,
                'registry-unauthenticated-pull': 'false',
                'id': lce['id'],
                'organization-id': module_org.id,
            }
        )
        docker_repo_uri = (
            f'{target_sat.hostname}/{pattern_prefix}-{content_view["label"]}/{docker_upstream_name}'
        ).lower()

        # 3. Try to pull in docker image from Satellite
        docker_pull_command = f'docker pull {docker_repo_uri}'
        result = module_container_contenthost.execute(docker_pull_command)
        assert result.status != 0

        # 4. Use Docker client to login to Satellite docker hub
        result = module_container_contenthost.execute(
            f'docker login -u {settings.server.admin_username}'
            f' -p {settings.server.admin_password} {target_sat.hostname}'
        )
        assert result.status == 0
        request.addfinalizer(
            lambda: module_container_contenthost.execute(f'docker logout {target_sat.hostname}')
        )

        # 5. Pull in docker image
        # publishing takes few seconds sometimes
        result, _ = wait_for(
            lambda: module_container_contenthost.execute(docker_pull_command),
            num_sec=60,
            delay=2,
            fail_condition=lambda out: out.status != 0,
            logger=logger,
        )
        assert result.status == 0

        # 6. Use Docker client to log out of Satellite docker hub
        result = module_container_contenthost.execute(f'docker logout {target_sat.hostname}')
        assert result.status == 0

        # 7. Try to pull in docker image
        result = module_container_contenthost.execute(docker_pull_command)
        assert result.status != 0

        # 8. Set 'Unauthenticated Pull' option to true
        target_sat.cli.LifecycleEnvironment.update(
            {
                'registry-unauthenticated-pull': 'true',
                'id': lce['id'],
                'organization-id': module_org.id,
            }
        )

        # 9. Pull in docker image
        result = module_container_contenthost.execute(docker_pull_command)
        assert result.status == 0

    def test_positive_pull_content_with_longer_name(
        self, request, target_sat, module_container_contenthost, module_org
    ):
        """Verify that long name CV publishes when CV & docker repo both have a larger name.

        :id: e0ac0be4-f5ff-4a88-bb29-33aa2d874f46

        :steps:

            1. Create Product, docker repo, CV and LCE with a long name
            2. Sync the repos
            3. Add repository to CV, Publish, and then Promote CV to LCE
            4. Pull in docker image

        :expectedresults:

            1. Long Product, repository, CV and LCE should create successfully
            2. Sync repository successfully
            3. Publish & Promote should success
            4. Can pull in docker images

        :BZ: 2127470

        :customerscenario: true
        """
        pattern_postfix = gen_string('alpha', 10).lower()

        product_name = f'containers-{pattern_postfix}'
        repo_name = f'repo-{pattern_postfix}'
        lce_name = f'lce-{pattern_postfix}'
        cv_name = f'cv-{pattern_postfix}'

        # 1. Create Product, docker repo, CV and LCE with a long name
        product = target_sat.cli_factory.make_product_wait(
            {'name': product_name, 'organization-id': module_org.id}
        )

        repo = _repo(
            target_sat,
            product['id'],
            name=repo_name,
            upstream_name=settings.container.upstream_name,
        )

        # 2. Sync the repos
        target_sat.cli.Repository.synchronize({'id': repo['id']})

        lce = target_sat.cli_factory.make_lifecycle_environment(
            {'name': lce_name, 'organization-id': module_org.id}
        )
        cv = target_sat.cli_factory.make_content_view(
            {'name': cv_name, 'composite': False, 'organization-id': module_org.id}
        )

        # 3. Add repository to CV, Publish, and then Promote CV to LCE
        target_sat.cli.ContentView.add_repository({'id': cv['id'], 'repository-id': repo['id']})

        target_sat.cli.ContentView.publish({'id': cv['id']})
        cv = target_sat.cli.ContentView.info({'id': cv['id']})
        target_sat.cli.ContentView.version_promote(
            {'id': cv['versions'][0]['id'], 'to-lifecycle-environment-id': lce['id']}
        )

        podman_pull_command = (
            f"podman pull --tls-verify=false {target_sat.hostname}/{module_org.label}"
            f"/{lce['label']}/{cv['label']}/{product['label']}/{repo_name}".lower()
        )

        # 4. Pull in docker image
        assert (
            module_container_contenthost.execute(
                f'podman login -u {settings.server.admin_username}'
                f' -p {settings.server.admin_password} {target_sat.hostname}'
            ).status
            == 0
        )
        request.addfinalizer(
            lambda: module_container_contenthost.execute(f'podman logout {target_sat.hostname}')
        )

        assert module_container_contenthost.execute(podman_pull_command).status == 0

    @pytest.fixture(scope='module')
    def stage_setup(
        self, module_target_sat, module_capsule_configured, module_org, module_lce, module_product
    ):
        """Setup for test_podman_cert_auth"""
        sat, caps = module_target_sat, module_capsule_configured

        # 1. Associate the organization and LCE to the capsule.
        res = sat.cli.Capsule.update({'name': caps.hostname, 'organization-ids': module_org.id})
        assert 'proxy updated' in str(res)

        caps.nailgun_capsule.content_add_lifecycle_environment(
            data={'environment_id': module_lce.id}
        )
        res = caps.nailgun_capsule.content_lifecycle_environments()
        assert len(res['results']) >= 1
        assert module_lce.id in [capsule_lce['id'] for capsule_lce in res['results']]

        # 2. Create and sync a docker repo.
        repo = _repo(sat, module_product.id, upstream_name='quay/busybox', url='https://quay.io')
        sat.cli.Repository.synchronize({'id': repo['id']})

        # 3. Create a CV with the repo, publish and promote it to a LCE, wait for capsule sync.
        cv = sat.cli_factory.make_content_view(
            {'organization-id': module_org.id, 'repository-ids': [repo['id']]}
        )
        timestamp = datetime.now(UTC)
        sat.cli.ContentView.publish({'id': cv['id']})
        cv = sat.cli.ContentView.info({'id': cv['id']})
        sat.cli.ContentView.version_promote(
            {'id': cv['versions'][0]['id'], 'to-lifecycle-environment-id': module_lce.id}
        )
        module_capsule_configured.wait_for_sync(start_time=timestamp)

        # 4. Create activation key for the LCE/CV.
        ak = sat.cli.ActivationKey.create(
            {
                'name': gen_string('alpha'),
                'organization-id': module_org.id,
                'lifecycle-environment-id': module_lce.id,
                'content-view-id': cv['id'],
            }
        )
        return Box(repo=repo, cv=cv, ak=ak)

    @pytest.mark.e2e
    @pytest.mark.parametrize('target_server', ['sat', 'caps'], ids=['satellite', 'capsule'])
    @pytest.mark.parametrize('gr_certs_setup', [False, True], ids=['manual-setup', 'GR-setup'])
    def test_podman_cert_auth(
        self,
        request,
        module_target_sat,
        module_capsule_configured,
        module_container_contenthost,
        stage_setup,
        target_server,
        gr_certs_setup,
        module_org,
        module_lce,
        module_product,
    ):
        """Verify the podman search and pull works with cert-based authentication for both,
        Satellite and Capsule, without need for login.

        :id: 7b1a457c-ae67-4a76-9f67-9074ea7f858a

        :parametrized: yes

        :Verifies: SAT-33254, SAT-33255, SAT-33260, SAT-39878

        :setup:
            1. Associate the organization and LCE to the capsule.
            2. Create and sync a docker repo.
            3. Create a CV with the repo, publish and promote it to a LCE, wait for capsule sync.
            4. Create activation key for the LCE/CV.

        :steps:
            1. Register a host to the LCE/CV environment.
            2. Configure podman certs for authentication (manual setup only).
            3. Try podman search all, ensure Library and repo images are not listed.
            4. Try podman search/pull for Library images, ensure it fails.
            5. Try podman search/pull for the LCE/CV, ensure it works.

        :expectedresults:
            1. Podman search/pull is restricted for Library (or any LCE missing in AK).
            2. Podman search/pull works for environments included in AK.
            3. The above applies for both, Satellite and Capsule.

        """
        server = module_capsule_configured if target_server == 'caps' else module_target_sat
        host = module_container_contenthost
        org, lce, prod = module_org, module_lce, module_product
        repo, cv, ak = stage_setup.repo, stage_setup.cv, stage_setup.ak

        # 1. Register a host to the LCE/CV environment.
        res = host.register(
            org, None, ak.name, server, force=True, setup_container_certs=gr_certs_setup
        )
        assert res.status == 0
        assert host.subscribed

        @request.addfinalizer
        def _finalize():
            host.unregister()
            host.delete_host_record()
            host.reset_podman_cert_auth(server)  # reset regardless how it was set

        # 2. Configure podman certs for authentication (manual setup only).
        if not gr_certs_setup:
            host.configure_podman_cert_auth(server)

        # 3. Try podman search all, ensure Library and repo images are not listed.
        org_prefix = f'{server.hostname}/{org.label}'
        lib_path = f'{org_prefix}/library'.lower()
        repo_path = f'{org_prefix}/{prod.label}/{repo.label}'.lower()
        cv_path = f'{org_prefix}/{lce.label}/{cv.label}/{prod.label}/{repo.label}'.lower()

        finds = host.execute(f'podman search {server.hostname}/').stdout
        assert lib_path not in finds
        assert repo_path not in finds
        assert cv_path in finds
        if not is_open('SAT-39878'):
            paths = [f.strip() for f in finds.split('\n') if 'NAME' not in f and len(f)]
            assert len(paths) == 1

        # 4. Try podman search/pull for Library images, ensure it fails.
        for path in [lib_path, repo_path]:
            assert host.execute(f'podman search {path}').stdout == ''
            assert host.execute(f'podman pull {path}').status

        # 5. Try podman search/pull for the LCE/CV, ensure it works.
        res = host.execute(f'podman search {cv_path}')
        assert cv_path in res.stdout
        res = host.execute(f'podman pull {cv_path}')
        assert res.status == 0
        request.addfinalizer(lambda: host.execute(f'podman rmi {cv_path}'))
        res = host.execute('podman images')
        assert cv_path in res.stdout

    def test_positive_revoke_registry_token_prevents_access(
        self, request, target_sat, function_product
    ):
        """Verify that revoking a registry access token prevents client access to registry.

        :id: 8b4e5c2a-1f3d-4a7e-9c8b-2d6f4e5a3b1c

        :steps:
            1. Sync a small container repo (quay/busybox from quay.io)
            2. Login to Satellite registry
            3. Pull the synced image - should succeed
            4. Revoke the registry personal access token via CLI
            5. Try to pull again - should fail with authentication error

        :expectedresults:
            After revoking the authentication token, podman pull fails with
            authentication error.

        :Verifies: SAT-38785
        """
        # 1. Sync a small container repo (quay/busybox from quay.io)
        repo = _repo(
            target_sat, function_product.id, upstream_name='quay/busybox', url='https://quay.io'
        )
        target_sat.cli.Repository.synchronize({'id': repo['id']})

        # Build the registry path for the synced image
        repo_info = target_sat.cli.Repository.info({'id': repo['id']})
        registry_path = repo_info['published-at']

        @request.addfinalizer
        def _cleanup():
            target_sat.execute(f'podman logout {target_sat.hostname}')
            target_sat.execute(f'podman rmi {registry_path}')

        # 2. Login to Satellite registry
        result = target_sat.execute(
            f'podman login -u {settings.server.admin_username}'
            f' -p {settings.server.admin_password} {target_sat.hostname}'
        )
        assert result.status == 0, f'Failed to login to registry: {result.stderr}'

        # 3. Pull the synced image - should succeed
        result = target_sat.execute(f'podman pull {registry_path}')
        assert result.status == 0, f'Failed to pull synced image: {result.stderr}'

        # 4. Revoke the registry personal access token via CLI
        admin_user = target_sat.cli.User.info({'login': settings.server.admin_username})
        target_sat.cli.User.access_token(
            action='revoke',
            options={'name': 'registry', 'user-id': admin_user['id']},
        )

        # 5. Try to pull again - should fail with authentication error
        result = target_sat.execute(f'podman pull {registry_path}')
        assert result.status != 0, 'Pull should have failed after token revocation'
        assert (
            'authentication required' in result.stderr.lower()
            or 'unauthorized' in result.stderr.lower()
        ), f'Expected authentication error, got: {result.stderr}'