deny endpoint access to system tracking feature based on license

awx/api/serializers.py

@@ -483,7 +483,7 @@ class BaseFactSerializer(BaseSerializer):
 
     def get_fields(self):
         ret = super(BaseFactSerializer, self).get_fields()
-        if 'module' in ret and feature_enabled('system_tracking'):
+        if 'module' in ret:
             # TODO: the values_list may pull in a LOT of entries before the distinct is called
             modules = Fact.objects.all().values_list('module', flat=True).distinct()
             choices = [(o, o.title()) for o in modules]

awx/api/views.py

@@ -1270,7 +1270,17 @@ class HostActivityStreamList(SubListAPIView):
         qs = self.request.user.get_queryset(self.model)
         return qs.filter(Q(host=parent) | Q(inventory=parent.inventory))
 
-class HostFactVersionsList(ListAPIView, ParentMixin):
+class SystemTrackingEnforcementMixin(APIView):
+    '''
+    Use check_permissions instead of initial() because it's in the OPTION's path as well
+    '''
+    def check_permissions(self, request):
+        if not feature_enabled("system_tracking"):
+            raise LicenseForbids("Your license does not permit use "
+                                 "of system tracking.")
+        return super(SystemTrackingEnforcementMixin, self).check_permissions(request)
+
+class HostFactVersionsList(ListAPIView, ParentMixin, SystemTrackingEnforcementMixin):
 
     model = Fact
     serializer_class = FactVersionSerializer
@@ -1278,10 +1288,6 @@ class HostFactVersionsList(ListAPIView, ParentMixin):
     new_in_220 = True
 
     def get_queryset(self):
-        if not feature_enabled("system_tracking"):
-            raise LicenseForbids("Your license does not permit use "
-                                 "of system tracking.")
-
         from_spec = self.request.query_params.get('from', None)
         to_spec = self.request.query_params.get('to', None)
         module_spec = self.request.query_params.get('module', None)
@@ -1299,7 +1305,7 @@ class HostFactVersionsList(ListAPIView, ParentMixin):
         queryset = self.get_queryset() or []
         return Response(dict(results=self.serializer_class(queryset, many=True).data))
 
-class HostFactCompareView(SubDetailAPIView):
+class HostFactCompareView(SubDetailAPIView, SystemTrackingEnforcementMixin):
 
     model = Fact
     new_in_220 = True
@@ -1307,11 +1313,6 @@ class HostFactCompareView(SubDetailAPIView):
     serializer_class = FactSerializer
 
     def retrieve(self, request, *args, **kwargs):
-        # Sanity check: Does the license allow system tracking?
-        if not feature_enabled('system_tracking'):
-            raise LicenseForbids('Your license does not permit use '
-                                 'of system tracking.')
-
         datetime_spec = request.query_params.get('datetime', None)
         module_spec = request.query_params.get('module', "ansible")
         datetime_actual = dateutil.parser.parse(datetime_spec) if datetime_spec is not None else now()

awx/main/tests/functional/api/test_fact_versions.py

@@ -16,6 +16,9 @@ from django.utils import timezone
 def mock_feature_enabled(feature, bypass_database=None):
     return True
 
+def mock_feature_disabled(feature, bypass_database=None):
+    return False
+
 def setup_common(hosts, fact_scans, get, user, epoch=timezone.now(), get_params={}, host_count=1):
     hosts = hosts(host_count=host_count)
     fact_scans(fact_scans=3, timestamp_epoch=epoch)
@@ -42,8 +45,33 @@ def check_response_facts(facts_known, response):
         assert timestamp_apiformat(fact_known.timestamp) == response.data['results'][i]['timestamp']
         check_url(response.data['results'][i]['related']['fact_view'], fact_known, fact_known.module)
 
+def check_system_tracking_feature_forbidden(response):
+    assert 402 == response.status_code
+    assert 'Your license does not permit use of system tracking.' == response.data['detail']
+
+@mock.patch('awx.api.views.feature_enabled', new=mock_feature_disabled)
+@pytest.mark.django_db
+@pytest.mark.license_feature
+def test_system_tracking_license_get(hosts, get, user):
+    hosts = hosts(host_count=1)
+    url = reverse('api:host_fact_versions_list', args=(hosts[0].pk,))
+    response = get(url, user('admin', True))
+
+    check_system_tracking_feature_forbidden(response)
+
+@mock.patch('awx.api.views.feature_enabled', new=mock_feature_disabled)
+@pytest.mark.django_db
+@pytest.mark.license_feature
+def test_system_tracking_license_options(hosts, options, user):
+    hosts = hosts(host_count=1)
+    url = reverse('api:host_fact_versions_list', args=(hosts[0].pk,))
+    response = options(url, None, user('admin', True))
+
+    check_system_tracking_feature_forbidden(response)
+
 @mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
 @pytest.mark.django_db
+@pytest.mark.license_feature
 def test_no_facts_db(hosts, get, user):
     hosts = hosts(host_count=1)
     url = reverse('api:host_fact_versions_list', args=(hosts[0].pk,))
@@ -72,6 +100,7 @@ def test_basic_fields(hosts, fact_scans, get, user):
 
 @mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
 @pytest.mark.django_db
+@pytest.mark.license_feature
 def test_basic_options_fields(hosts, fact_scans, options, user):
     hosts = hosts(host_count=1)
     fact_scans(fact_scans=1)

awx/main/tests/functional/api/test_fact_view.py

@@ -9,6 +9,9 @@ from django.utils import timezone
 def mock_feature_enabled(feature, bypass_database=None):
     return True
 
+def mock_feature_disabled(feature, bypass_database=None):
+    return False
+
 # TODO: Consider making the fact_scan() fixture a Class, instead of a function, and move this method into it
 def find_fact(facts, host_id, module_name, timestamp):
     for f in facts:
@@ -26,6 +29,30 @@ def setup_common(hosts, fact_scans, get, user, epoch=timezone.now(), module_name
     fact_known = find_fact(facts, hosts[0].id, module_name, epoch)
     return (fact_known, response)
 
+def check_system_tracking_feature_forbidden(response):
+    assert 402 == response.status_code
+    assert 'Your license does not permit use of system tracking.' == response.data['detail']
+
+@mock.patch('awx.api.views.feature_enabled', new=mock_feature_disabled)
+@pytest.mark.django_db
+@pytest.mark.license_feature
+def test_system_tracking_license_get(hosts, get, user):
+    hosts = hosts(host_count=1)
+    url = reverse('api:host_fact_compare_view', args=(hosts[0].pk,))
+    response = get(url, user('admin', True))
+
+    check_system_tracking_feature_forbidden(response)
+
+@mock.patch('awx.api.views.feature_enabled', new=mock_feature_disabled)
+@pytest.mark.django_db
+@pytest.mark.license_feature
+def test_system_tracking_license_options(hosts, options, user):
+    hosts = hosts(host_count=1)
+    url = reverse('api:host_fact_compare_view', args=(hosts[0].pk,))
+    response = options(url, None, user('admin', True))
+
+    check_system_tracking_feature_forbidden(response)
+
 @mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
 @pytest.mark.django_db
 def test_no_fact_found(hosts, get, user):

awx/main/tests/functional/commands/test_cleanup_facts.py

@@ -19,6 +19,9 @@ from awx.main.models.inventory import Host
 def mock_feature_enabled(feature, bypass_database=None):
     return True
 
+def mock_feature_disabled(feature, bypass_database=None):
+    return False
+
 @pytest.mark.django_db
 def test_cleanup_granularity(fact_scans, hosts):
     epoch = timezone.now()
@@ -92,6 +95,15 @@ def test_cleanup_logic(fact_scans, hosts):
             timestamp_pivot -= granularity
             assert fact.timestamp == timestamp_pivot
 
+@mock.patch('awx.main.management.commands.cleanup_facts.feature_enabled', new=mock_feature_disabled)
+@pytest.mark.django_db
+@pytest.mark.license_feature
+def test_system_tracking_feature_disabled(mocker):
+    cmd = Command()
+    with pytest.raises(CommandError) as err:
+        cmd.handle(None)
+    assert 'The System Tracking feature is not enabled for your Tower instance' in err.value
+
 @mock.patch('awx.main.management.commands.cleanup_facts.feature_enabled', new=mock_feature_enabled)
 @pytest.mark.django_db
 def test_parameters_ok(mocker):

pytest.ini

@@ -6,3 +6,4 @@ python_files = *.py
 addopts = --reuse-db
 markers =
     ac: access control test
+    license_feature: ensure license features are accessible or not depending on license