Incident management: improve validation

api/debian/changelog

@@ -1,3 +1,9 @@
+ooni-api (1.0.65) unstable; urgency=medium
+
+  * https://github.com/ooni/backend/issues/707
+
+ -- Federico Ceratto <federico@debian.org>  Thu, 24 Aug 2023 16:06:21 +0200
+
 ooni-api (1.0.64) unstable; urgency=medium
 
   * Set "mine" and "archived" as bools

api/ooniapi/incidents.py

@@ -27,10 +27,10 @@
 )
 from ooniapi.config import metrics
 from ooniapi.database import query_click, optimize_table, insert_click, raw_query
-from ooniapi.errors import BaseOONIException, jerror, NoProposedChanges
+from ooniapi.errors import BaseOONIException, jerror
 from ooniapi.errors import OwnershipPermissionError, InvalidRequest
 from ooniapi.urlparams import param_bool
-from ooniapi.utils import nocachejson, cachedjson, generate_random_intuid
+from ooniapi.utils import nocachejson, generate_random_intuid
 
 log: logging.Logger
 
@@ -166,14 +166,23 @@ def prepare_incident_dict(d: dict):
         log.debug(f"Invalid incident update request. Keys: {sorted(d)}")
         raise InvalidRequest()
 
-    d["published"] = int(d.get("published", 0))
     d["start_time"] = datetime.strptime(d["start_time"], "%Y-%m-%dT%H:%M:%SZ")
     if d["end_time"] is not None:
         d["end_time"] = datetime.strptime(d["end_time"], "%Y-%m-%dT%H:%M:%SZ")
+        delta = d["end_time"] - d["start_time"]
+        if delta.total_seconds() < 0:
+            raise InvalidRequest()
+
     if not d["title"] or not d["text"]:
         log.debug("Invalid incident update request: empty title or desc")
         raise InvalidRequest()
 
+    try:
+        for asn in d["ASNs"]:
+            int(asn)
+    except Exception:
+        raise InvalidRequest()
+
 
 def user_cannot_update(incident_id: str) -> bool:
     # Check if there is already an incident and belogs to a different user
@@ -265,6 +274,8 @@ def post_update_incident(action: str) -> Response:
         if req is None:
             raise InvalidRequest()
 
+        req["published"] = int(req.get("published", 0))
+
         if action in ("update", "delete"):
             incident_id = req.get("id")
             if incident_id is None:
@@ -274,16 +285,19 @@ def post_update_incident(action: str) -> Response:
         if action == "create":
             incident_id = str(generate_random_intuid(current_app))
             req["id"] = incident_id
-            if get_client_role() != "admin":
-                req["published"] = 0  # users cannot publish
+            if get_client_role() != "admin" and req["published"] == 1:
+                raise InvalidRequest
+
             log.info(f"Creating incident {incident_id}")
 
         elif action == "update":
             if get_client_role() != "admin":
                 if user_cannot_update(incident_id):
                     raise OwnershipPermissionError
-                req["published"] = 0  # users cannot publish
-                log.info(f"Updating incident {incident_id}")
+                if req["published"] == 1:
+                    raise InvalidRequest
+
+            log.info(f"Updating incident {incident_id}")
 
         elif action == "delete":
             if get_client_role() != "admin":

api/tests/integ/test_incidents.py

@@ -154,7 +154,6 @@ def test_crud_general(cleanup, client, adminsession, usersession):
 
 
 def test_crud_user_create(cleanup, client, adminsession, usersession):
-    # Create. Users attempts to publish the incident but is prevented.
     title = "integ-test-2"
     new = dict(
         start_time=datetime(2020, 1, 1),
@@ -163,7 +162,7 @@ def test_crud_user_create(cleanup, client, adminsession, usersession):
         title=title,
         text="foo bar\nbaz\n",
         event_type="incident",
-        published=True,
+        published=False,
         CCs=["UK", "FR"],
         test_names=["web_connectivity"],
         ASNs=[1, 2],
@@ -206,6 +205,79 @@ def test_crud_user_create(cleanup, client, adminsession, usersession):
     assert i == expected
 
 
+def test_crud_user_create_cannot_publish(cleanup, client, adminsession, usersession):
+    # Create. Users attempts to publish the incident but is prevented.
+    title = "integ-test-3"
+    new = dict(
+        start_time=datetime(2020, 1, 1),
+        end_time=None,
+        reported_by="ooni",
+        title=title,
+        text="foo bar\nbaz\n",
+        event_type="incident",
+        published=True,
+        CCs=["UK", "FR"],
+        test_names=["web_connectivity"],
+        ASNs=[1, 2],
+        domains=[],
+        tags=["integ-test"],
+        links=[
+            "https://explorer.ooni.org/chart/mat?test_name=web_connectivity&axis_x=measurement_start_day&since=2023-04-16&until=2023-05-16&time_grain=day"
+        ],
+    )
+    d = dict(**new)
+    r = usersession.post("/api/v1/incidents/create", json=d)
+    assert r.status_code == 400, r.json
+
+
+def test_crud_user_create_invalid_asns(cleanup, client, adminsession, usersession):
+    title = "integ-test-4"
+    new = dict(
+        start_time=datetime(2020, 1, 1),
+        end_time=None,
+        reported_by="ooni",
+        title=title,
+        text="foo bar\nbaz\n",
+        event_type="incident",
+        published=False,
+        CCs=["UK", "FR"],
+        test_names=["web_connectivity"],
+        ASNs=[1, 2, "foo"],
+        domains=[],
+        tags=["integ-test"],
+        links=[
+            "https://explorer.ooni.org/chart/mat?test_name=web_connectivity&axis_x=measurement_start_day&since=2023-04-16&until=2023-05-16&time_grain=day"
+        ],
+    )
+    d = dict(**new)
+    r = usersession.post("/api/v1/incidents/create", json=d)
+    assert r.status_code == 400, r.json
+
+
+def test_crud_user_create_invalid_dates(cleanup, client, adminsession, usersession):
+    title = "integ-test-5"
+    new = dict(
+        start_time=datetime(2020, 1, 1),
+        end_time=datetime(2019, 1, 1),
+        reported_by="ooni",
+        title=title,
+        text="foo bar\nbaz\n",
+        event_type="incident",
+        published=False,
+        CCs=["UK", "FR"],
+        test_names=["web_connectivity"],
+        ASNs=[1, 2],
+        domains=[],
+        tags=["integ-test"],
+        links=[
+            "https://explorer.ooni.org/chart/mat?test_name=web_connectivity&axis_x=measurement_start_day&since=2023-04-16&until=2023-05-16&time_grain=day"
+        ],
+    )
+    d = dict(**new)
+    r = usersession.post("/api/v1/incidents/create", json=d)
+    assert r.status_code == 400, r.json
+
+
 def test_crud_invalid_fields(client, adminsession, usersession):
     # Create
     new = dict(