Prevent launching ad hoc commands when license has expired.

awx/api/permissions.py

@@ -108,7 +108,7 @@ class ModelAccessPermission(permissions.BasePermission):
             raise PermissionDenied('your account is inactive')
 
         # Always allow superusers (as long as they are active).
-        if request.user.is_superuser:
+        if getattr(view, 'always_allow_superuser', True) and request.user.is_superuser:
             return True
 
         # Check if view supports the request method before checking permission

awx/api/views.py

@@ -2309,6 +2309,7 @@ class AdHocCommandList(ListCreateAPIView):
     model = AdHocCommand
     serializer_class = AdHocCommandListSerializer
     new_in_220 = True
+    always_allow_superuser = False
 
     @csrf_exempt
     @transaction.non_atomic_requests

awx/main/access.py

@@ -2,6 +2,7 @@
 # All Rights Reserved.
 
 # Python
+import os
 import sys
 import logging
 
@@ -147,7 +148,7 @@ class BaseAccess(object):
     def check_license(self, add_host=False):
         reader = TaskSerializer()
         validation_info = reader.from_file()
-        if 'test' in sys.argv or 'jenkins' in sys.argv:
+        if ('test' in sys.argv or 'jenkins' in sys.argv) and not os.environ.get('SKIP_LICENSE_FIXUP_FOR_TEST', ''):
             validation_info['free_instances'] = 99999999
             validation_info['time_remaining'] = 99999999
             validation_info['grace_period_remaining'] = 99999999

awx/main/tests/ad_hoc.py

@@ -6,6 +6,7 @@ import glob
 import os
 import subprocess
 import tempfile
+import time
 import mock
 
 # Django
@@ -568,6 +569,13 @@ class AdHocCommandApiTest(BaseAdHocCommandTest):
         with self.current_user('admin'):
             response = self.run_test_ad_hoc_command(become_enabled=True)
             self.assertEqual(response['become_enabled'], True)
+        
+        # Try to run with expired license.
+        self.create_expired_license_file()
+        with self.current_user('admin'):
+            self.run_test_ad_hoc_command(expect=403)
+        with self.current_user('normal'):
+            self.run_test_ad_hoc_command(expect=403)
 
     @mock.patch('awx.main.tasks.BaseTask.run_pexpect', side_effect=run_pexpect_mock)
     def test_ad_hoc_command_detail(self, ignore):
@@ -748,6 +756,13 @@ class AdHocCommandApiTest(BaseAdHocCommandTest):
             self.assertEqual(response['passwords_needed_to_start'], [])
             response = self.post(url, {}, expect=400)
 
+        # Try to relaunch with expired license.
+        with self.current_user('admin'):
+            response = self.run_test_ad_hoc_command(inventory=self.inventory2.pk)
+        self.create_expired_license_file()
+        with self.current_user('admin'):
+            self.post(response['related']['relaunch'], {}, expect=403)
+
     def test_ad_hoc_command_events_list(self):
         # TODO: Create test events instead of relying on playbooks execution
 
@@ -1049,6 +1064,13 @@ class AdHocCommandApiTest(BaseAdHocCommandTest):
             response = self.get(inventory_url, expect=200)
             self.assertTrue(response['can_run_ad_hoc_commands'])
 
+        # Try to run with expired license.
+        self.create_expired_license_file()
+        with self.current_user('admin'):
+            self.run_test_ad_hoc_command(url=url, expect=403)
+        with self.current_user('normal'):
+            self.run_test_ad_hoc_command(url=url, expect=403)
+
     def test_host_ad_hoc_commands_list(self):
         # TODO: Figure out why this test needs pexpect
 
@@ -1100,6 +1122,13 @@ class AdHocCommandApiTest(BaseAdHocCommandTest):
             self.patch(url, {}, expect=401)
             self.delete(url, expect=401)
 
+        # Try to run with expired license.
+        self.create_expired_license_file()
+        with self.current_user('admin'):
+            self.run_test_ad_hoc_command(url=url, expect=403)
+        with self.current_user('normal'):
+            self.run_test_ad_hoc_command(url=url, expect=403)
+
     def test_group_ad_hoc_commands_list(self):
         # TODO: Figure out why this test needs pexpect
 
@@ -1156,6 +1185,13 @@ class AdHocCommandApiTest(BaseAdHocCommandTest):
             self.patch(url, {}, expect=401)
             self.delete(url, expect=401)
 
+        # Try to run with expired license.
+        self.create_expired_license_file()
+        with self.current_user('admin'):
+            self.run_test_ad_hoc_command(url=url, expect=403)
+        with self.current_user('normal'):
+            self.run_test_ad_hoc_command(url=url, expect=403)
+
     def test_host_ad_hoc_command_events_list(self):
         # TODO: Mock run_pexpect. Create test events instead of relying on playbooks execution
 

awx/main/tests/base.py

@@ -186,6 +186,13 @@ class BaseTestMixin(QueueTestMixin, MockCommonlySlowTestMixin):
         self._temp_paths.append(license_path)
         os.environ['AWX_LICENSE_FILE'] = license_path
 
+    def create_expired_license_file(self, instance_count=1000, grace_period=False):
+        license_date = time.time() - 1
+        if not grace_period:
+            license_date -= 2592000
+        self.create_test_license_file(instance_count, license_date)
+        os.environ['SKIP_LICENSE_FIXUP_FOR_TEST'] = '1'
+
     def assertElapsedLessThan(self, seconds):
         elapsed = time.time() - self._start_time
         self.assertTrue(elapsed < seconds, 'elapsed time of %0.3fs is greater than %0.3fs' % (elapsed, seconds))