diff --git a/product_portfolio/admin.py b/product_portfolio/admin.py index 498620ca..b69f5ca0 100644 --- a/product_portfolio/admin.py +++ b/product_portfolio/admin.py @@ -7,7 +7,6 @@ # import copy -import uuid from django import forms from django.contrib import admin @@ -59,8 +58,6 @@ from product_portfolio.importers import CodebaseResourceImporter from product_portfolio.importers import ProductComponentImporter from product_portfolio.importers import ProductPackageImporter -from product_portfolio.importers import paused_product_package_reevaluation -from product_portfolio.importers import reevaluate_products from product_portfolio.inlines import CodebaseResourceUsageDeployedFromInline from product_portfolio.inlines import CodebaseResourceUsageDeployedToInline from product_portfolio.inlines import ProductComponentInline @@ -74,7 +71,6 @@ from product_portfolio.models import ProductRelationStatus from product_portfolio.models import ProductStatus from reporting.filters import ReportingQueryListFilter -from vulnerabilities.triage.models import ProductTriageRuleset @admin.register(ProductStatus, site=dejacode_site) @@ -230,28 +226,11 @@ def has_delete_permission(self, request, obj=None): return super().has_delete_permission(request, obj) def save_model(self, request, obj, form, change): - """ - Add view/change/delete Object permissions to the Product creator. - - Add the copy of the ProductComponent and ProductPackage during a `save_as` since - those value are not in the form as inlines. - """ + """Add view/change/delete Object permissions to the Product creator.""" super().save_model(request, obj, form, change) if not change: assign_all_object_permissions(request.user, obj) - if "_saveasnew" in request.POST: - old_product_id = request.resolver_match.kwargs.get("object_id") - old_product = self.get_object(request, old_product_id) - with paused_product_package_reevaluation(): - for model_class in [ProductComponent, ProductPackage, ProductTriageRuleset]: - for relationship in model_class.objects.filter(product=old_product): - relationship.id = None - relationship.uuid = uuid.uuid4() - relationship.product = obj - relationship.save() - reevaluate_products([obj]) - def get_obj_perms_user_select_form(self, request): """ Return a UserManageForm class using the request object for User Dataspace scoping. @@ -390,6 +369,7 @@ class ProductAdmin( ProductPackageInline, ] form = ProductAdminForm + save_as = False actions = ["evaluate_policy_rules"] actions_to_remove = ["copy_to", "compare_with", "delete_selected"] navigation_buttons = True diff --git a/product_portfolio/api.py b/product_portfolio/api.py index c98cfd5e..56196c98 100644 --- a/product_portfolio/api.py +++ b/product_portfolio/api.py @@ -562,8 +562,9 @@ def clone(self, request, *args, **kwargs): Clone this Product into a new one. Always copies the base Product fields under the given name/version. - Optionally copies the inventory (Components, Packages, and Dependencies), - Codebase resources, Vulnerability triage rules, and object Permissions, + Optionally copies the inventory (Components, Packages, Dependencies, and + vulnerability analyses), Codebase resources and deployment links, Vulnerability + triage rules, and object Permissions, depending on the submitted flags (all default to true). """ product = self.get_object() diff --git a/product_portfolio/forms.py b/product_portfolio/forms.py index 5cce1068..971f495a 100644 --- a/product_portfolio/forms.py +++ b/product_portfolio/forms.py @@ -13,6 +13,7 @@ from django.conf import settings from django.core.exceptions import ValidationError from django.db import transaction +from django.db.models import Q from django.forms import BaseModelFormSet from django.forms.formsets import DELETION_FIELD_NAME from django.urls import reverse_lazy @@ -56,6 +57,7 @@ from dje.widgets import AwesompleteInputWidget from dje.widgets import DatePicker from product_portfolio.models import CodebaseResource +from product_portfolio.models import CodebaseResourceUsage from product_portfolio.models import Product from product_portfolio.models import ProductComponent from product_portfolio.models import ProductComponentAssignedLicense @@ -65,6 +67,7 @@ from product_portfolio.models import ScanCodeProject from product_portfolio.tasks import pull_project_data_from_scancodeio_task from product_portfolio.tasks import scancodeio_submit_project_task +from vulnerabilities.models import VulnerabilityAnalysis from vulnerabilities.triage.models import ProductTriageRuleset @@ -133,13 +136,7 @@ class ProductForm( DataspacedModelForm, ): default_on_addition_fields = ["configuration_status"] - save_as = True - clone_m2m_classes = [ - ProductComponent, - ProductPackage, - CodebaseResource, - ProductTriageRuleset, - ] + save_as = False keywords = KeywordsField() @@ -196,11 +193,6 @@ def assign_object_perms(self, user): if user.has_perm("product_portfolio.delete_product"): assign_perm("delete_product", user, self.instance) - def _clone_m2m(self, original_instance_id, cloned_instance): - cloned_pairs_by_model = super()._clone_m2m(original_instance_id, cloned_instance) - relink_cloned_codebase_resources(cloned_pairs_by_model) - return cloned_pairs_by_model - @property def helper(self): helper = super().helper @@ -221,7 +213,6 @@ def helper(self): Group("vulnerabilities_risk_threshold", HTML(""), HTML("")), HTML("
"), Submit("submit", self.submit_label, css_class="btn-success"), - self.save_as_new_submit, ), ) @@ -341,6 +332,52 @@ def _clone_relations(self, model_class, target_product): return cloned_pairs + def _clone_vulnerability_analyses(self, cloned_pairs_by_model, target_product): + cloned_package_by_original_pk = dict(cloned_pairs_by_model.get(ProductPackage, [])) + if not cloned_package_by_original_pk: + return + + for analysis in VulnerabilityAnalysis.objects.filter(product=self.source_product): + cloned_product_package = cloned_package_by_original_pk.get(analysis.product_package_id) + if not cloned_product_package: + continue + + VulnerabilityAnalysis.objects.create( + dataspace=target_product.dataspace, + product_package=cloned_product_package, + vulnerability=analysis.vulnerability, + state=analysis.state, + justification=analysis.justification, + responses=analysis.responses, + is_reachable=analysis.is_reachable, + detail=analysis.detail, + created_by=self.user, + last_modified_by=self.user, + ) + + def _clone_deployment_links(self, cloned_pairs_by_model, target_product): + cloned_resource_by_original_pk = dict(cloned_pairs_by_model.get(CodebaseResource, [])) + if not cloned_resource_by_original_pk: + return + + original_resource_ids = set(cloned_resource_by_original_pk) + usages = CodebaseResourceUsage.objects.filter( + Q(deployed_from_id__in=original_resource_ids) + | Q(deployed_to_id__in=original_resource_ids) + ) + for usage in usages: + deployed_from = cloned_resource_by_original_pk.get(usage.deployed_from_id) + deployed_to = cloned_resource_by_original_pk.get(usage.deployed_to_id) + if not deployed_from or not deployed_to: + continue + + CodebaseResourceUsage.objects.create( + uuid=uuid.uuid4(), + dataspace=target_product.dataspace, + deployed_from=deployed_from, + deployed_to=deployed_to, + ) + def save(self, commit=True): from product_portfolio.importers import paused_product_package_reevaluation from product_portfolio.importers import reevaluate_products @@ -359,8 +396,14 @@ def save(self, commit=True): cloned_pairs = self._clone_relations(model_class, instance) cloned_pairs_by_model[model_class] = cloned_pairs + if self.cleaned_data.get("copy_inventory"): + self._clone_vulnerability_analyses(cloned_pairs_by_model, instance) + relink_cloned_codebase_resources(cloned_pairs_by_model) + if self.cleaned_data.get("copy_codebase_resources"): + self._clone_deployment_links(cloned_pairs_by_model, instance) + reevaluate_products([instance]) if self.cleaned_data.get("copy_object_permissions"): diff --git a/product_portfolio/tests/test_admin.py b/product_portfolio/tests/test_admin.py index cce8a4ef..be3e54bd 100644 --- a/product_portfolio/tests/test_admin.py +++ b/product_portfolio/tests/test_admin.py @@ -25,8 +25,6 @@ from product_portfolio.models import ProductComponent from product_portfolio.models import ProductPackage from product_portfolio.tests import make_product_dependency -from vulnerabilities.triage.models import ProductTriageRuleset -from vulnerabilities.triage.models import TriageRuleset class ProductPortfolioAdminsTestCase(TestCase): @@ -178,16 +176,6 @@ def test_product_admin_changeform_unique_validation(self): expected = {NON_FIELD_ERRORS: [error_msg]} self.assertEqual(expected, response.context_data["adminform"].form.errors) - url = self.product1.get_admin_url() - data["_saveasnew"] = "Save as new" - response = self.client.post(url, data) - self.assertContains(response, error_msg) - self.assertEqual(expected, response.context_data["adminform"].form.errors) - - data["version"] = "1.0" - response = self.client.post(url, data) - self.assertEqual(response.status_code, 302) - def test_productcomponent_admin_changeform_feature_datalist(self): self.client.login(username=self.user.username, password="secret") url = reverse("admin:product_portfolio_productcomponent_add") @@ -423,72 +411,13 @@ def test_productcomponent_admin_check_for_update_actions(self): self.assertIn(expected1, actions) self.assertIn(expected2, actions) - def test_product_admin_changeform_save_as_includes_relationships(self): - self.client.login(username=self.user.username, password="secret") - - ProductComponent.objects.create( - product=self.product1, component=self.component1, dataspace=self.dataspace - ) - ProductPackage.objects.create( - product=self.product1, package=self.package1, dataspace=self.dataspace - ) - ruleset = TriageRuleset.objects.create( - name="Upgrade Ruleset", precedence=100, dataspace=self.dataspace - ) - ProductTriageRuleset.objects.create( - product=self.product1, ruleset=ruleset, dataspace=self.dataspace - ) + def test_product_admin_changeform_does_not_expose_save_as_new(self): + self.client.force_login(self.user) url = self.product1.get_admin_url() - data = { - "_saveasnew": "Save as new", - "name": self.product1.name, - "version": "new version", - "productcomponents-INITIAL_FORMS": 0, - "productcomponents-TOTAL_FORMS": 0, - "productpackages-INITIAL_FORMS": 0, - "productpackages-TOTAL_FORMS": 0, - } - - response = self.client.post(url, data) - self.assertEqual(response.status_code, 302) - new_product = Product.unsecured_objects.get(name=self.product1.name, version="new version") - self.assertEqual(1, new_product.productcomponents.count()) - self.assertEqual(1, new_product.productpackages.count()) - self.assertEqual(1, new_product.product_triage_rulesets.count()) - - @patch("product_portfolio.admin.reevaluate_products") - def test_product_admin_changeform_save_as_evaluates_the_new_product_once(self, mock_reevaluate): - # Cloning must evaluate the new product exactly once, not once per cloned - # ProductPackage relationship. - package2 = make_package(self.dataspace) - ProductPackage.objects.create( - product=self.product1, package=self.package1, dataspace=self.dataspace - ) - ProductPackage.objects.create( - product=self.product1, package=package2, dataspace=self.dataspace - ) - self.client.login(username=self.user.username, password="secret") - - url = self.product1.get_admin_url() - data = { - "_saveasnew": "Save as new", - "name": self.product1.name, - "version": "new version 2", - "productcomponents-INITIAL_FORMS": 0, - "productcomponents-TOTAL_FORMS": 0, - "productpackages-INITIAL_FORMS": 0, - "productpackages-TOTAL_FORMS": 0, - } - - response = self.client.post(url, data) - - self.assertEqual(response.status_code, 302) - new_product = Product.unsecured_objects.get( - name=self.product1.name, version="new version 2" - ) - self.assertEqual(2, new_product.productpackages.count()) - mock_reevaluate.assert_called_once_with([new_product]) + response = self.client.get(url) + self.assertNotContains(response, 'name="_saveasnew"') + self.assertNotContains(response, "Save as new") def test_codebaseresource_admin_changeform_product_prefill_on_save_addanother(self): self.client.login(username=self.user.username, password="secret") diff --git a/product_portfolio/tests/test_api.py b/product_portfolio/tests/test_api.py index d4989709..4870adc2 100644 --- a/product_portfolio/tests/test_api.py +++ b/product_portfolio/tests/test_api.py @@ -583,8 +583,26 @@ def test_api_product_endpoint_import_manifests_action(self): def test_api_product_endpoint_clone_action(self): url = reverse("api_v2:product-clone", args=[self.product1.uuid]) - make_product_package(self.product1) + product_package = make_product_package(self.product1) make_product_triage_ruleset(self.product1) + vulnerability = make_vulnerability( + self.product1.dataspace, affecting=[product_package.package] + ) + source_analysis = make_vulnerability_analysis(product_package, vulnerability) + deployed_from = CodebaseResource.objects.create( + product=self.product1, path="/src/", dataspace=self.product1.dataspace + ) + deployed_to = CodebaseResource.objects.create( + product=self.product1, + path="/deploy/", + is_deployment_path=True, + dataspace=self.product1.dataspace, + ) + source_usage = CodebaseResourceUsage.objects.create( + dataspace=self.product1.dataspace, + deployed_from=deployed_from, + deployed_to=deployed_to, + ) self.client.login(username=self.base_user.username, password="secret") response = self.client.get(url) @@ -615,6 +633,15 @@ def test_api_product_endpoint_clone_action(self): self.assertEqual(str(cloned_product.uuid), response.data["uuid"]) self.assertEqual(1, cloned_product.productpackages.count()) self.assertEqual(1, cloned_product.product_triage_rulesets.count()) + cloned_analysis = cloned_product.vulnerability_analyses.get() + self.assertEqual(cloned_product.productpackages.get(), cloned_analysis.product_package) + self.assertEqual(source_analysis.state, cloned_analysis.state) + self.assertEqual(source_analysis.justification, cloned_analysis.justification) + self.assertEqual(source_analysis.responses, cloned_analysis.responses) + self.assertEqual(source_analysis.detail, cloned_analysis.detail) + cloned_usage = CodebaseResourceUsage.objects.get(deployed_from__product=cloned_product) + self.assertNotEqual(source_usage.deployed_from_id, cloned_usage.deployed_from_id) + self.assertNotEqual(source_usage.deployed_to_id, cloned_usage.deployed_to_id) # Submitting the same name/version again is rejected response = self.client.post(url, data) @@ -625,6 +652,7 @@ def test_api_product_endpoint_clone_action(self): "name": self.product1.name, "version": "3.0", "copy_inventory": False, + "copy_codebase_resources": False, "copy_triage_rulesets": False, } response = self.client.post(url, data) @@ -632,6 +660,11 @@ def test_api_product_endpoint_clone_action(self): empty_clone = Product.objects.get_queryset(self.base_user).get(version="3.0") self.assertEqual(0, empty_clone.productpackages.count()) self.assertEqual(0, empty_clone.product_triage_rulesets.count()) + self.assertEqual(0, empty_clone.vulnerability_analyses.count()) + self.assertEqual(0, empty_clone.codebaseresources.count()) + self.assertEqual( + 0, CodebaseResourceUsage.objects.filter(deployed_from__product=empty_clone).count() + ) def test_api_product_endpoint_import_from_scan_action(self): url = reverse("api_v2:product-import-from-scan", args=[self.product1.uuid]) diff --git a/product_portfolio/tests/test_views.py b/product_portfolio/tests/test_views.py index 11e653e7..80bc3cbf 100644 --- a/product_portfolio/tests/test_views.py +++ b/product_portfolio/tests/test_views.py @@ -49,6 +49,7 @@ from product_portfolio.forms import ProductGridConfigurationForm from product_portfolio.forms import ProductPackageForm from product_portfolio.models import CodebaseResource +from product_portfolio.models import CodebaseResourceUsage from product_portfolio.models import Product from product_portfolio.models import ProductComponent from product_portfolio.models import ProductItemPurpose @@ -2165,74 +2166,35 @@ def test_product_portfolio_product_update_view_is_active_false(self): expected = f"Product "{self.product1}" was successfully updated." self.assertContains(response, expected) - def test_product_portfolio_product_update_save_as_new(self): - self.client.login(username=self.super_user.username, password="secret") - - self.assertTrue(self.super_user.is_superuser) + def test_product_portfolio_product_update_does_not_expose_save_as_new(self): + self.client.force_login(self.super_user) add_url = reverse("product_portfolio:product_add") response = self.client.get(add_url) expected = 'value="Save as new"' self.assertNotContains(response, expected) - self.super_user.is_superuser = False - self.super_user.is_staff = False - self.super_user.save() - self.super_user = add_perms(self.super_user, ["change_product"]) - assign_perm("change_product", self.super_user, self.product1) change_url = self.product1.get_change_url() response = self.client.get(change_url) self.assertNotContains(response, expected) + self.assertNotContains(response, 'name="save_as_new"') - self.super_user = add_perms(self.super_user, ["add_product"]) - response = self.client.get(change_url) - self.assertContains(response, expected) - - product_component = ProductComponent.objects.create( - product=self.product1, component=self.component1, dataspace=self.dataspace - ) - product_package = ProductPackage.objects.create( + @patch("product_portfolio.importers.reevaluate_products") + def test_product_portfolio_product_clone_view_reevaluates_once(self, mock_reevaluate): + self.client.force_login(self.basic_user) + assign_perm("change_product", self.basic_user, self.product1) + self.basic_user = add_perms(self.basic_user, ["add_product"]) + ProductPackage.objects.create( product=self.product1, package=self.package1, dataspace=self.dataspace ) - CodebaseResource.objects.create( - path="/path1/", - product=self.product1, - product_component=product_component, - product_package=product_package, - dataspace=self.dataspace, - ) - ruleset = TriageRuleset.objects.create( - name="Upgrade Ruleset", precedence=100, dataspace=self.dataspace - ) - ProductTriageRuleset.objects.create( - product=self.product1, ruleset=ruleset, dataspace=self.dataspace - ) - initial_product_count = Product.objects.get_queryset(self.super_user).count() - data = { - "name": self.product1.name, - "version": "new version", - "is_active": "on", - "save_as_new": "Save as new", - } - - response = self.client.post(change_url, data, follow=True) - - new_count = Product.objects.get_queryset(self.super_user).count() - self.assertEqual(new_count, initial_product_count + 1) - cloned_product = Product.objects.get_queryset(self.super_user).latest("id") - self.assertRedirects(response, cloned_product.get_absolute_url()) - self.assertContains(response, "was successfully cloned.") - - self.assertNotEqual(self.product1.id, cloned_product.id) - self.assertEqual(1, cloned_product.productcomponents.count()) - self.assertEqual(1, cloned_product.productpackages.count()) - self.assertEqual(1, cloned_product.codebaseresources.count()) - self.assertEqual(1, cloned_product.product_triage_rulesets.count()) + response = self.client.post( + self.product1.get_clone_url(), + {"name": self.product1.name, "version": "2.0", "copy_inventory": "on"}, + ) - cloned_resource = cloned_product.codebaseresources.get() - self.assertEqual(cloned_product.productcomponents.get(), cloned_resource.product_component) - self.assertEqual(cloned_product.productpackages.get(), cloned_resource.product_package) - cloned_resource.full_clean() + self.assertEqual(302, response.status_code) + cloned_product = Product.unsecured_objects.get(name=self.product1.name, version="2.0") + mock_reevaluate.assert_called_once_with([cloned_product]) def test_product_portfolio_product_clone_view(self): clone_url = self.product1.get_clone_url() @@ -2259,12 +2221,19 @@ def test_product_portfolio_product_clone_view(self): license_expression=license1.key, dataspace=self.dataspace, ) - ProductPackage.objects.create( + product_package = ProductPackage.objects.create( product=self.product1, package=self.package1, license_expression=license1.key, dataspace=self.dataspace, ) + vulnerability = make_vulnerability(self.dataspace, affecting=[self.package1]) + source_analysis = make_vulnerability_analysis( + product_package, + vulnerability, + is_reachable=True, + state=VulnerabilityAnalysis.State.EXPLOITABLE, + ) CodebaseResource.objects.create( path="/path1/", product=self.product1, dataspace=self.dataspace ) @@ -2308,6 +2277,14 @@ def test_product_portfolio_product_clone_view(self): cloned_productpackage = cloned_product.productpackages.get() self.assertEqual(license1.key, cloned_productpackage.license_expression) self.assertEqual([license1], list(cloned_productpackage.licenses.all())) + cloned_analysis = cloned_product.vulnerability_analyses.get() + self.assertEqual(cloned_productpackage, cloned_analysis.product_package) + self.assertEqual(source_analysis.state, cloned_analysis.state) + self.assertEqual(source_analysis.justification, cloned_analysis.justification) + self.assertEqual(source_analysis.responses, cloned_analysis.responses) + self.assertEqual(source_analysis.is_reachable, cloned_analysis.is_reachable) + self.assertEqual(source_analysis.detail, cloned_analysis.detail) + self.assertNotEqual(source_analysis.pk, cloned_analysis.pk) # Submitting the same name/version again is rejected. response = self.client.post(clone_url, data) @@ -2328,6 +2305,7 @@ def test_product_portfolio_product_clone_view(self): self.assertEqual(0, empty_clone.codebaseresources.count()) self.assertEqual(0, empty_clone.dependencies.count()) self.assertEqual(0, empty_clone.product_triage_rulesets.count()) + self.assertEqual(0, empty_clone.vulnerability_analyses.count()) self.assertNotIn("view_product", get_user_perms(self.super_user, empty_clone)) def test_product_portfolio_product_clone_view_relinks_codebase_resources(self): @@ -2340,13 +2318,24 @@ def test_product_portfolio_product_clone_view_relinks_codebase_resources(self): product_package = ProductPackage.objects.create( product=self.product1, package=self.package1, dataspace=self.dataspace ) - CodebaseResource.objects.create( + deployed_from = CodebaseResource.objects.create( path="/path1/", product=self.product1, product_component=product_component, product_package=product_package, dataspace=self.dataspace, ) + deployed_to = CodebaseResource.objects.create( + path="/deploy/path1/", + product=self.product1, + is_deployment_path=True, + dataspace=self.dataspace, + ) + source_usage = CodebaseResourceUsage.objects.create( + dataspace=self.dataspace, + deployed_from=deployed_from, + deployed_to=deployed_to, + ) data = { "name": self.product1.name, @@ -2356,25 +2345,42 @@ def test_product_portfolio_product_clone_view_relinks_codebase_resources(self): } self.client.post(clone_url, data) cloned_product = Product.unsecured_objects.get(name=self.product1.name, version="2.0") - cloned_resource = cloned_product.codebaseresources.get() + cloned_resource = cloned_product.codebaseresources.get(path="/path1/") self.assertEqual(cloned_product.productcomponents.get(), cloned_resource.product_component) self.assertEqual(cloned_product.productpackages.get(), cloned_resource.product_package) cloned_resource.full_clean() + cloned_usage = CodebaseResourceUsage.objects.get(deployed_from__product=cloned_product) + self.assertEqual(cloned_product, cloned_usage.deployed_from.product) + self.assertEqual(cloned_product, cloned_usage.deployed_to.product) + self.assertNotEqual(source_usage.deployed_from_id, cloned_usage.deployed_from_id) + self.assertNotEqual(source_usage.deployed_to_id, cloned_usage.deployed_to_id) # Without the inventory, the links to the source product relations are cleared. data["version"] = "3.0" data["copy_inventory"] = "" self.client.post(clone_url, data) cloned_product = Product.unsecured_objects.get(name=self.product1.name, version="3.0") - cloned_resource = cloned_product.codebaseresources.get() + cloned_resource = cloned_product.codebaseresources.get(path="/path1/") self.assertIsNone(cloned_resource.product_component) self.assertIsNone(cloned_resource.product_package) cloned_resource.full_clean() - source_resource = self.product1.codebaseresources.get() + source_resource = self.product1.codebaseresources.get(path="/path1/") self.assertEqual(product_component, source_resource.product_component) self.assertEqual(product_package, source_resource.product_package) + data["version"] = "4.0" + data["copy_codebase_resources"] = "" + self.client.post(clone_url, data) + empty_resource_clone = Product.unsecured_objects.get(name=self.product1.name, version="4.0") + self.assertEqual(0, empty_resource_clone.codebaseresources.count()) + self.assertEqual( + 0, + CodebaseResourceUsage.objects.filter( + deployed_from__product=empty_resource_clone + ).count(), + ) + def test_product_portfolio_product_delete_view(self): delete_url = self.product1.get_delete_url() details_url = self.product1.get_absolute_url()