feat: add scopes for alert actions

alerta/models/enums.py

@@ -62,6 +62,13 @@ class Scope(str):
     admin_alerta = 'admin:alerta'
     read_alerts = 'read:alerts'
     write_alerts = 'write:alerts'
+    write_alerts_actions = 'write:alerts.actions'
+    write_alerts_actions_close = 'write:alerts.actions.close'
+    write_alerts_actions_open = 'write:alerts.actions.open'
+    write_alerts_actions_shelve = 'write:alerts.actions.shelve'
+    write_alerts_actions_unshelve = 'write:alerts.actions.unshelve'
+    write_alerts_actions_ack = 'write:alerts.actions.ack'
+    write_alerts_actions_unack = 'write:alerts.actions.unack'
     delete_alerts = 'delete:alerts'
     admin_alerts = 'admin:alerts'
     read_blackouts = 'read:blackouts'

alerta/models/permission.py

@@ -97,17 +97,23 @@ def is_in_scope(cls, want_scope: str, have_scopes: List[Scope]) -> bool:
             return True
         if want_scope in have_scopes or want_scope.split(':')[0] in have_scopes or want_scope.split('.')[0] in have_scopes:
             return True
-        elif want_scope.startswith('read'):
+        if '.' in want_scope:
+            s = want_scope.split('.')
+            b = s.pop(0)
+            for a in s:
+                b = f"{b}.{a}"
+                if b in have_scopes:
+                    return True
+        if want_scope.startswith('read'):
             return cls.is_in_scope(want_scope.replace('read', 'write'), have_scopes)
-        elif want_scope.startswith('write'):
+        if want_scope.startswith('write'):
             return cls.is_in_scope(want_scope.replace('write', 'admin'), have_scopes)
-        elif want_scope.startswith('delete'):
+        if want_scope.startswith('delete'):
             if want_scope in current_app.config['DELETE_SCOPES']:
                 return cls.is_in_scope(want_scope.replace('delete', 'admin'), have_scopes)
             else:
                 return cls.is_in_scope(want_scope.replace('delete', 'write'), have_scopes)
-        else:
-            return False
+        return False
 
     @classmethod
     def lookup(cls, login: str, roles: List[str]) -> List[Scope]:

alerta/views/alerts.py

@@ -177,14 +177,7 @@ def set_status(alert_id):
         raise ApiError('failed to set status', 500)
 
 
-# action alert
-@api.route('/alert/<alert_id>/action', methods=['OPTIONS', 'PUT'])
-@cross_origin()
-@permission(Scope.write_alerts)
-@timer(status_timer)
-@jsonp
-def action_alert(alert_id):
-    action = request.json.get('action', None)
+def handle_action(alert_id, action):
     text = request.json.get('text', f'{action} operator action')
     timeout = request.json.get('timeout', None)
 
@@ -227,15 +220,81 @@ def action_alert(alert_id):
         raise ApiError('failed to action alert', 500)
 
 
-# action alert
-@api.route('/alerts/action', methods=['OPTIONS', 'PUT'])
+@api.route('/alert/<alert_id>/action/ack', methods=['OPTIONS', 'PUT'])
+@cross_origin()
+@permission(Scope.write_alerts_actions_ack)
+@timer(status_timer)
+@jsonp
+def ack_alert(alert_id):
+    return handle_action(alert_id, 'ack')
+
+
+@api.route('/alert/<alert_id>/action/unack', methods=['OPTIONS', 'PUT'])
+@cross_origin()
+@permission(Scope.write_alerts_actions_unack)
+@timer(status_timer)
+@jsonp
+def unack_alert(alert_id):
+    return handle_action(alert_id, 'unack')
+
+
+@api.route('/alert/<alert_id>/action/shelve', methods=['OPTIONS', 'PUT'])
+@cross_origin()
+@permission(Scope.write_alerts_actions_shelve)
+@timer(status_timer)
+@jsonp
+def shelve_alert(alert_id):
+    return handle_action(alert_id, 'shelve')
+
+
+@api.route('/alert/<alert_id>/action/unshelve', methods=['OPTIONS', 'PUT'])
+@cross_origin()
+@permission(Scope.write_alerts_actions_unshelve)
+@timer(status_timer)
+@jsonp
+def unshelve_alert(alert_id):
+    return handle_action(alert_id, 'unshelve')
+
+
+# @api.route('/alert/<alert_id>/action/delete', methods=['OPTIONS', 'PUT'])
+# @cross_origin()
+# @permission(Scope.write_alerts_actions_delete)
+# @timer(status_timer)
+# @jsonp
+# def delete_alert(alert_id):
+#     return handle_action(alert_id, 'delete')
+
+
+@api.route('/alert/<alert_id>/action/open', methods=['OPTIONS', 'PUT'])
+@cross_origin()
+@permission(Scope.write_alerts_actions_open)
+@timer(status_timer)
+@jsonp
+def open_alert(alert_id):
+    return handle_action(alert_id, 'open')
+
+
+@api.route('/alert/<alert_id>/action/close', methods=['OPTIONS', 'PUT'])
+@cross_origin()
+@permission(Scope.write_alerts_actions_close)
+@timer(status_timer)
+@jsonp
+def close_alert(alert_id):
+    return handle_action(alert_id, 'close')
+
+
+@api.route('/alert/<alert_id>/action', methods=['OPTIONS', 'PUT'])
 @cross_origin()
 @permission(Scope.write_alerts)
 @timer(status_timer)
 @jsonp
-def action_alerts():
-    alert_ids = request.json.get('alerts')
+def action_alert(alert_id):
     action = request.json.get('action', None)
+    return handle_action(alert_id, action)
+
+
+def actions(action: str):
+    alert_ids = request.json.get('alerts')
     text = request.json.get('text', f'{action} operator action')
     timeout = request.json.get('timeout', None)
 
@@ -277,6 +336,72 @@ def post(alerts: list[Alert], action, text, timeout):
     else:
         raise ApiError('failed to action alert', 500)
 
+# action alert
+
+
+@api.route('/alerts/action', methods=['OPTIONS', 'PUT'])
+@cross_origin()
+@permission(Scope.write_alerts)
+@timer(status_timer)
+@jsonp
+def action_alerts():
+    action = request.json.get('action', None)
+    return actions(action)
+
+
+@api.route('/alerts/action/ack', methods=['OPTIONS', 'PUT'])
+@cross_origin()
+@permission(Scope.write_alerts_actions_ack)
+@timer(status_timer)
+@jsonp
+def ack_alerts():
+    return actions('ack')
+
+
+@api.route('/alerts/action/unack', methods=['OPTIONS', 'PUT'])
+@cross_origin()
+@permission(Scope.write_alerts_actions_unack)
+@timer(status_timer)
+@jsonp
+def unack_alerts():
+    return actions('unack')
+
+
+@api.route('/alerts/action/close', methods=['OPTIONS', 'PUT'])
+@cross_origin()
+@permission(Scope.write_alerts_actions_close)
+@timer(status_timer)
+@jsonp
+def close_alerts():
+    return actions('close')
+
+
+@api.route('/alerts/action/open', methods=['OPTIONS', 'PUT'])
+@cross_origin()
+@permission(Scope.write_alerts_actions_open)
+@timer(status_timer)
+@jsonp
+def open_alerts():
+    return actions('open')
+
+
+@api.route('/alerts/action/shelve', methods=['OPTIONS', 'PUT'])
+@cross_origin()
+@permission(Scope.write_alerts_actions_shelve)
+@timer(status_timer)
+@jsonp
+def shelve_alerts():
+    return actions('shelve')
+
+
+@api.route('/alerts/action/unshelve', methods=['OPTIONS', 'PUT'])
+@cross_origin()
+@permission(Scope.write_alerts_actions_unshelve)
+@timer(status_timer)
+@jsonp
+def unshelve_alerts():
+    return actions('unshelve')
+
 
 # tag
 @api.route('/alert/<alert_id>/tag', methods=['OPTIONS', 'PUT'])

tests/test_actions.py

@@ -146,6 +146,55 @@ def test_alert(self):
         self.assertEqual(data['alert']['severity'], 'major')
         self.assertEqual(data['alert']['status'], alarm_model.DEFAULT_STATUS)
 
+    def test_alert_action(self):
+
+        # create alert
+        response = self.client.post('/alert', data=json.dumps(self.major_alert), headers=self.headers)
+        self.assertEqual(response.status_code, 201)
+        data = json.loads(response.data.decode('utf-8'))
+        self.assertEqual(data['alert']['resource'], self.resource)
+        self.assertEqual(data['alert']['status'], 'open')
+        self.assertEqual(data['alert']['duplicateCount'], 0)
+        self.assertEqual(data['alert']['severity'], 'major')
+
+        alert_id = data['id']
+
+        # ack alert
+        response = self.client.put('/alert/' + alert_id + '/action/ack', data='{}', headers=self.headers)
+        self.assertEqual(response.status_code, 200)
+        response = self.client.get('/alert/' + alert_id)
+        self.assertEqual(response.status_code, 200)
+        data = json.loads(response.data.decode('utf-8'))
+        self.assertEqual(data['alert']['severity'], 'major')
+        self.assertEqual(data['alert']['status'], 'ack')
+
+        # un-ack alert
+        response = self.client.put('/alert/' + alert_id + '/action/unack', data='{}', headers=self.headers)
+        self.assertEqual(response.status_code, 200)
+        response = self.client.get('/alert/' + alert_id)
+        self.assertEqual(response.status_code, 200)
+        data = json.loads(response.data.decode('utf-8'))
+        self.assertEqual(data['alert']['severity'], 'major')
+        self.assertEqual(data['alert']['status'], 'open')
+
+        # close alert
+        response = self.client.put('/alert/' + alert_id + '/action/close', data='{}', headers=self.headers)
+        self.assertEqual(response.status_code, 200)
+        response = self.client.get('/alert/' + alert_id)
+        self.assertEqual(response.status_code, 200)
+        data = json.loads(response.data.decode('utf-8'))
+        self.assertEqual(data['alert']['severity'], 'normal')
+        self.assertEqual(data['alert']['status'], 'closed')
+
+        # open alert
+        response = self.client.put('/alert/' + alert_id + '/action/open', data='{}', headers=self.headers)
+        self.assertEqual(response.status_code, 200)
+        response = self.client.get('/alert/' + alert_id)
+        self.assertEqual(response.status_code, 200)
+        data = json.loads(response.data.decode('utf-8'))
+        self.assertEqual(data['alert']['severity'], 'major')
+        self.assertEqual(data['alert']['status'], alarm_model.DEFAULT_STATUS)
+
     def test_alerts(self):
 
         # create alert
@@ -237,6 +286,97 @@ def test_alerts(self):
         self.assertEqual(data['alert']['severity'], 'major')
         self.assertEqual(data['alert']['status'], alarm_model.DEFAULT_STATUS)
 
+    def test_alerts_action(self):
+
+        # create alert
+        response = self.client.post('/alerts', data=json.dumps([{**self.major_alert, 'resource': '1'}, {**self.major_alert, 'resource': '2'}]), headers=self.headers)
+        self.assertEqual(response.status_code, 201)
+        data = json.loads(response.data.decode('utf-8'))
+        first, second = data['alerts']
+
+        self.assertEqual(first['resource'], '1')
+        self.assertEqual(first['status'], 'open')
+        self.assertEqual(first['duplicateCount'], 0)
+
+        self.assertEqual(second['resource'], '2')
+        self.assertEqual(second['status'], 'open')
+        self.assertEqual(second['duplicateCount'], 0)
+
+        first_id = first['id']
+        second_id = second['id']
+
+        # ack alert
+        response = self.client.put('/alerts/action/ack',
+                                   data=json.dumps({'alerts': [first_id, second_id]}),
+                                   headers=self.headers)
+        self.assertEqual(response.status_code, 200)
+
+        response = self.client.get('/alert/' + first_id)
+        self.assertEqual(response.status_code, 200)
+        data = json.loads(response.data.decode('utf-8'))
+        self.assertEqual(data['alert']['severity'], 'major')
+        self.assertEqual(data['alert']['status'], 'ack')
+
+        response = self.client.get('/alert/' + second_id)
+        self.assertEqual(response.status_code, 200)
+        data = json.loads(response.data.decode('utf-8'))
+        self.assertEqual(data['alert']['severity'], 'major')
+        self.assertEqual(data['alert']['status'], 'ack')
+
+        # un-ack alert
+        response = self.client.put('/alerts/action/unack',
+                                   data=json.dumps({'alerts': [first_id, second_id],}),
+                                   headers=self.headers)
+        self.assertEqual(response.status_code, 200)
+
+        response = self.client.get('/alert/' + first_id)
+        self.assertEqual(response.status_code, 200)
+        data = json.loads(response.data.decode('utf-8'))
+        self.assertEqual(data['alert']['severity'], 'major')
+        self.assertEqual(data['alert']['status'], 'open')
+
+        response = self.client.get('/alert/' + second_id)
+        self.assertEqual(response.status_code, 200)
+        data = json.loads(response.data.decode('utf-8'))
+        self.assertEqual(data['alert']['severity'], 'major')
+        self.assertEqual(data['alert']['status'], 'open')
+
+        # close alert
+        response = self.client.put('/alerts/action/close',
+                                   data=json.dumps({'alerts': [first_id, second_id],}),
+                                   headers=self.headers)
+        self.assertEqual(response.status_code, 200)
+
+        response = self.client.get('/alert/' + first_id)
+        self.assertEqual(response.status_code, 200)
+        data = json.loads(response.data.decode('utf-8'))
+        self.assertEqual(data['alert']['severity'], 'normal')
+        self.assertEqual(data['alert']['status'], 'closed')
+
+        response = self.client.get('/alert/' + second_id)
+        self.assertEqual(response.status_code, 200)
+        data = json.loads(response.data.decode('utf-8'))
+        self.assertEqual(data['alert']['severity'], 'normal')
+        self.assertEqual(data['alert']['status'], 'closed')
+
+        # open alert
+        response = self.client.put('/alerts/action/open',
+                                   data=json.dumps({'alerts': [first_id, second_id]}),
+                                   headers=self.headers)
+        self.assertEqual(response.status_code, 200)
+
+        response = self.client.get('/alert/' + first_id)
+        self.assertEqual(response.status_code, 200)
+        data = json.loads(response.data.decode('utf-8'))
+        self.assertEqual(data['alert']['severity'], 'major')
+        self.assertEqual(data['alert']['status'], alarm_model.DEFAULT_STATUS)
+
+        response = self.client.get('/alert/' + second_id)
+        self.assertEqual(response.status_code, 200)
+        data = json.loads(response.data.decode('utf-8'))
+        self.assertEqual(data['alert']['severity'], 'major')
+        self.assertEqual(data['alert']['status'], alarm_model.DEFAULT_STATUS)
+
     def test_unwind_actions(self):
 
         # new alert => open