Add privilege check in triggers and streams (#200)

tests/e2e/streams/CMakeLists.txt

@@ -6,6 +6,9 @@ add_custom_target(memgraph__e2e__streams__${FILE_NAME} ALL
         DEPENDS ${CMAKE_CURRENT_SOURCE_DIR}/${FILE_NAME})
 endfunction()
 
+copy_streams_e2e_python_files(common.py)
+copy_streams_e2e_python_files(conftest.py)
 copy_streams_e2e_python_files(streams_tests.py)
+copy_streams_e2e_python_files(streams_owner_tests.py)
 copy_streams_e2e_python_files(streams_test_runner.sh)
 add_subdirectory(transformations)

tests/e2e/streams/common.py

@@ -0,0 +1,110 @@
+import mgclient
+import time
+
+# These are the indices of the different values in the result of SHOW STREAM
+# query
+NAME = 0
+TOPICS = 1
+CONSUMER_GROUP = 2
+BATCH_INTERVAL = 3
+BATCH_SIZE = 4
+TRANSFORM = 5
+OWNER = 6
+IS_RUNNING = 7
+
+
+def execute_and_fetch_all(cursor, query):
+    cursor.execute(query)
+    return cursor.fetchall()
+
+
+def connect(**kwargs):
+    connection = mgclient.connect(host="localhost", port=7687, **kwargs)
+    connection.autocommit = True
+    return connection
+
+
+def timed_wait(fun):
+    start_time = time.time()
+    seconds = 10
+
+    while True:
+        current_time = time.time()
+        elapsed_time = current_time - start_time
+
+        if elapsed_time > seconds:
+            return False
+
+        if fun():
+            return True
+
+        time.sleep(0.1)
+
+
+def check_one_result_row(cursor, query):
+    start_time = time.time()
+    seconds = 10
+
+    while True:
+        current_time = time.time()
+        elapsed_time = current_time - start_time
+
+        if elapsed_time > seconds:
+            return False
+
+        cursor.execute(query)
+        results = cursor.fetchall()
+        if len(results) < 1:
+            time.sleep(0.1)
+            continue
+
+        return len(results) == 1
+
+
+def check_vertex_exists_with_topic_and_payload(cursor, topic, payload_bytes):
+    assert check_one_result_row(cursor,
+                                "MATCH (n: MESSAGE {"
+                                f"payload: '{payload_bytes.decode('utf-8')}',"
+                                f"topic: '{topic}'"
+                                "}) RETURN n")
+
+
+def get_stream_info(cursor, stream_name):
+    stream_infos = execute_and_fetch_all(cursor, "SHOW STREAMS")
+    for stream_info in stream_infos:
+        if (stream_info[NAME] == stream_name):
+            return stream_info
+
+    return None
+
+
+def get_is_running(cursor, stream_name):
+    stream_info = get_stream_info(cursor, stream_name)
+
+    assert stream_info
+    return stream_info[IS_RUNNING]
+
+
+def start_stream(cursor, stream_name):
+    execute_and_fetch_all(cursor, f"START STREAM {stream_name}")
+
+    assert get_is_running(cursor, stream_name)
+
+
+def stop_stream(cursor, stream_name):
+    execute_and_fetch_all(cursor, f"STOP STREAM {stream_name}")
+
+    assert not get_is_running(cursor, stream_name)
+
+
+def drop_stream(cursor, stream_name):
+    execute_and_fetch_all(cursor, f"DROP STREAM {stream_name}")
+
+    assert get_stream_info(cursor, stream_name) is None
+
+
+def check_stream_info(cursor, stream_name, expected_stream_info):
+    stream_info = get_stream_info(cursor, stream_name)
+    assert len(stream_info) == len(expected_stream_info)
+    for info, expected_info in zip(stream_info, expected_stream_info):
+        assert info == expected_info

tests/e2e/streams/conftest.py

@@ -0,0 +1,45 @@
+import pytest
+from kafka import KafkaProducer
+from kafka.admin import KafkaAdminClient, NewTopic
+
+from common import execute_and_fetch_all, connect, NAME
+
+# To run these test locally a running Kafka sever is necessery. The test tries
+# to connect on localhost:9092.
+
+
+@pytest.fixture(autouse=True)
+def connection():
+    connection = connect()
+    yield connection
+    cursor = connection.cursor()
+    execute_and_fetch_all(cursor, "MATCH (n) DETACH DELETE n")
+    stream_infos = execute_and_fetch_all(cursor, "SHOW STREAMS")
+    for stream_info in stream_infos:
+        execute_and_fetch_all(cursor, f"DROP STREAM {stream_info[NAME]}")
+    users = execute_and_fetch_all(cursor, "SHOW USERS")
+    for username, in users:
+        execute_and_fetch_all(cursor, f"DROP USER {username}")
+
+
+@pytest.fixture(scope="function")
+def topics():
+    admin_client = KafkaAdminClient(
+        bootstrap_servers="localhost:9092", client_id='test')
+
+    topics = []
+    topics_to_create = []
+    for index in range(3):
+        topic = f"topic_{index}"
+        topics.append(topic)
+        topics_to_create.append(NewTopic(name=topic,
+                                num_partitions=1, replication_factor=1))
+
+    admin_client.create_topics(new_topics=topics_to_create, timeout_ms=5000)
+    yield topics
+    admin_client.delete_topics(topics=topics, timeout_ms=5000)
+
+
+@pytest.fixture(scope="function")
+def producer():
+    yield KafkaProducer(bootstrap_servers="localhost:9092")

tests/e2e/streams/streams_owner_tests.py

@@ -0,0 +1,151 @@
+import sys
+import pytest
+import time
+import mgclient
+import common
+
+
+def get_cursor_with_user(username):
+    connection = common.connect(username=username, password="")
+    return connection.cursor()
+
+
+def create_admin_user(cursor, admin_user):
+    common.execute_and_fetch_all(cursor, f"CREATE USER {admin_user}")
+    common.execute_and_fetch_all(
+        cursor, f"GRANT ALL PRIVILEGES TO {admin_user}")
+
+
+def create_stream_user(cursor, stream_user):
+    common.execute_and_fetch_all(cursor, f"CREATE USER {stream_user}")
+    common.execute_and_fetch_all(
+        cursor, f"GRANT STREAM TO {stream_user}")
+
+
+def test_ownerless_stream(producer, topics, connection):
+    assert len(topics) > 0
+    userless_cursor = connection.cursor()
+    common.execute_and_fetch_all(userless_cursor,
+                                 "CREATE STREAM ownerless "
+                                 f"TOPICS {topics[0]} "
+                                 f"TRANSFORM transform.simple")
+    common.start_stream(userless_cursor, "ownerless")
+    time.sleep(1)
+
+    admin_user = "admin_user"
+    create_admin_user(userless_cursor, admin_user)
+
+    producer.send(topics[0], b"first message").get(timeout=60)
+    assert common.timed_wait(
+        lambda: not common.get_is_running(userless_cursor, "ownerless"))
+
+    assert len(common.execute_and_fetch_all(
+        userless_cursor, "MATCH (n) RETURN n")) == 0
+
+    common.execute_and_fetch_all(userless_cursor, f"DROP USER {admin_user}")
+    common.start_stream(userless_cursor, "ownerless")
+    time.sleep(1)
+
+    second_message = b"second message"
+    producer.send(topics[0], second_message).get(timeout=60)
+    common.check_vertex_exists_with_topic_and_payload(
+        userless_cursor, topics[0], second_message)
+
+    assert len(common.execute_and_fetch_all(
+        userless_cursor, "MATCH (n) RETURN n")) == 1
+
+
+def test_owner_is_shown(topics, connection):
+    assert len(topics) > 0
+    userless_cursor = connection.cursor()
+
+    stream_user = "stream_user"
+    create_stream_user(userless_cursor, stream_user)
+    stream_cursor = get_cursor_with_user(stream_user)
+
+    common.execute_and_fetch_all(stream_cursor, "CREATE STREAM test "
+                                 f"TOPICS {topics[0]} "
+                                 f"TRANSFORM transform.simple")
+
+    common.check_stream_info(userless_cursor, "test", ("test", [
+        topics[0]], "mg_consumer", None, None,
+        "transform.simple", stream_user, False))
+
+
+def test_insufficient_privileges(producer, topics, connection):
+    assert len(topics) > 0
+    userless_cursor = connection.cursor()
+
+    admin_user = "admin_user"
+    create_admin_user(userless_cursor, admin_user)
+    admin_cursor = get_cursor_with_user(admin_user)
+
+    stream_user = "stream_user"
+    create_stream_user(userless_cursor, stream_user)
+    stream_cursor = get_cursor_with_user(stream_user)
+
+    common.execute_and_fetch_all(stream_cursor,
+                                 "CREATE STREAM insufficient_test "
+                                 f"TOPICS {topics[0]} "
+                                 f"TRANSFORM transform.simple")
+
+    # the stream is started by admin, but should check against the owner
+    # privileges
+    common.start_stream(admin_cursor, "insufficient_test")
+    time.sleep(1)
+
+    producer.send(topics[0], b"first message").get(timeout=60)
+    assert common.timed_wait(
+        lambda: not common.get_is_running(userless_cursor, "insufficient_test"))
+
+    assert len(common.execute_and_fetch_all(
+        userless_cursor, "MATCH (n) RETURN n")) == 0
+
+    common.execute_and_fetch_all(
+        admin_cursor, f"GRANT CREATE TO {stream_user}")
+    common.start_stream(userless_cursor, "insufficient_test")
+    time.sleep(1)
+
+    second_message = b"second message"
+    producer.send(topics[0], second_message).get(timeout=60)
+    common.check_vertex_exists_with_topic_and_payload(
+        userless_cursor, topics[0], second_message)
+
+    assert len(common.execute_and_fetch_all(
+        userless_cursor, "MATCH (n) RETURN n")) == 1
+
+
+def test_happy_case(producer, topics, connection):
+    assert len(topics) > 0
+    userless_cursor = connection.cursor()
+
+    admin_user = "admin_user"
+    create_admin_user(userless_cursor, admin_user)
+    admin_cursor = get_cursor_with_user(admin_user)
+
+    stream_user = "stream_user"
+    create_stream_user(userless_cursor, stream_user)
+    stream_cursor = get_cursor_with_user(stream_user)
+    common.execute_and_fetch_all(
+        admin_cursor, f"GRANT CREATE TO {stream_user}")
+
+    common.execute_and_fetch_all(stream_cursor,
+                                 "CREATE STREAM insufficient_test "
+                                 f"TOPICS {topics[0]} "
+                                 f"TRANSFORM transform.simple")
+
+    common.start_stream(stream_cursor, "insufficient_test")
+    time.sleep(1)
+
+    first_message = b"first message"
+    producer.send(topics[0], first_message).get(timeout=60)
+
+    common.check_vertex_exists_with_topic_and_payload(
+        userless_cursor, topics[0], first_message)
+
+    assert len(common.execute_and_fetch_all(
+        userless_cursor, "MATCH (n) RETURN n")) == 1
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/streams/streams_test_runner.sh

@@ -3,4 +3,4 @@
 # This workaround is necessary to run in the same virtualenv as the e2e runner.py
 
 DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
-python3 "$DIR/streams_tests.py"
+python3 "$DIR/$1"

tests/e2e/streams/streams_tests.py

@@ -1,28 +1,11 @@
 #!/usr/bin/python3
 
-# To run these test locally a running Kafka sever is necessery. The test tries
-# to connect on localhost:9092.
-
-# All tests are implemented in this file, because using the same test fixtures
-# in multiple files is not possible in a straightforward way
-
 import sys
 import pytest
 import mgclient
 import time
 from multiprocessing import Process, Value
-from kafka import KafkaProducer
-from kafka.admin import KafkaAdminClient, NewTopic
-
-# These are the indices of the different values in the result of SHOW STREAM
-# query
-NAME = 0
-TOPICS = 1
-CONSUMER_GROUP = 2
-BATCH_INTERVAL = 3
-BATCH_SIZE = 4
-TRANSFORM = 5
-IS_RUNNING = 6
+import common
 
 # These are the indices of the query and parameters in the result of CHECK
 # STREAM query
@@ -35,155 +18,22 @@ TRANSFORMATIONS_TO_CHECK = [
 SIMPLE_MSG = b'message'
 
 
-def execute_and_fetch_all(cursor, query):
-    cursor.execute(query)
-    return cursor.fetchall()
-
-
-def connect():
-    connection = mgclient.connect(host="localhost", port=7687)
-    connection.autocommit = True
-    return connection
-
-
-@pytest.fixture(autouse=True)
-def connection():
-    connection = connect()
-    yield connection
-    cursor = connection.cursor()
-    execute_and_fetch_all(cursor, "MATCH (n) DETACH DELETE n")
-    stream_infos = execute_and_fetch_all(cursor, "SHOW STREAMS")
-    for stream_info in stream_infos:
-        execute_and_fetch_all(cursor, f"DROP STREAM {stream_info[NAME]}")
-
-
-@pytest.fixture(scope="function")
-def topics():
-    admin_client = KafkaAdminClient(
-        bootstrap_servers="localhost:9092", client_id='test')
-
-    topics = []
-    topics_to_create = []
-    for index in range(3):
-        topic = f"topic_{index}"
-        topics.append(topic)
-        topics_to_create.append(NewTopic(name=topic,
-                                num_partitions=1, replication_factor=1))
-
-    admin_client.create_topics(new_topics=topics_to_create, timeout_ms=5000)
-    yield topics
-    admin_client.delete_topics(topics=topics, timeout_ms=5000)
-
-
-@pytest.fixture(scope="function")
-def producer():
-    yield KafkaProducer(bootstrap_servers="localhost:9092")
-
-
-def timed_wait(fun):
-    start_time = time.time()
-    seconds = 10
-
-    while True:
-        current_time = time.time()
-        elapsed_time = current_time - start_time
-
-        if elapsed_time > seconds:
-            return False
-
-        if fun():
-            return True
-
-
-def check_one_result_row(cursor, query):
-    start_time = time.time()
-    seconds = 10
-
-    while True:
-        current_time = time.time()
-        elapsed_time = current_time - start_time
-
-        if elapsed_time > seconds:
-            return False
-
-        cursor.execute(query)
-        results = cursor.fetchall()
-        if len(results) < 1:
-            time.sleep(0.1)
-            continue
-
-        return len(results) == 1
-
-
-def check_vertex_exists_with_topic_and_payload(cursor, topic, payload_bytes):
-    assert check_one_result_row(cursor,
-                                "MATCH (n: MESSAGE {"
-                                f"payload: '{payload_bytes.decode('utf-8')}',"
-                                f"topic: '{topic}'"
-                                "}) RETURN n")
-
-
-def get_stream_info(cursor, stream_name):
-    stream_infos = execute_and_fetch_all(cursor, "SHOW STREAMS")
-    for stream_info in stream_infos:
-        if (stream_info[NAME] == stream_name):
-            return stream_info
-
-    return None
-
-
-def get_is_running(cursor, stream_name):
-    stream_info = get_stream_info(cursor, stream_name)
-
-    assert stream_info
-    return stream_info[IS_RUNNING]
-
-
-def start_stream(cursor, stream_name):
-    execute_and_fetch_all(cursor, f"START STREAM {stream_name}")
-
-    assert get_is_running(cursor, stream_name)
-
-
-def stop_stream(cursor, stream_name):
-    execute_and_fetch_all(cursor, f"STOP STREAM {stream_name}")
-
-    assert not get_is_running(cursor, stream_name)
-
-
-def drop_stream(cursor, stream_name):
-    execute_and_fetch_all(cursor, f"DROP STREAM {stream_name}")
-
-    assert get_stream_info(cursor, stream_name) is None
-
-
-def check_stream_info(cursor, stream_name, expected_stream_info):
-    stream_info = get_stream_info(cursor, stream_name)
-    assert len(stream_info) == len(expected_stream_info)
-    for info, expected_info in zip(stream_info, expected_stream_info):
-        assert info == expected_info
-
-##############################################
-# Tests
-##############################################
-
-
 @pytest.mark.parametrize("transformation", TRANSFORMATIONS_TO_CHECK)
 def test_simple(producer, topics, connection, transformation):
     assert len(topics) > 0
     cursor = connection.cursor()
-    execute_and_fetch_all(cursor,
-                          "CREATE STREAM test "
-                          f"TOPICS {','.join(topics)} "
-                          f"TRANSFORM {transformation}")
-    start_stream(cursor, "test")
+    common.execute_and_fetch_all(cursor,
+                                 "CREATE STREAM test "
+                                 f"TOPICS {','.join(topics)} "
+                                 f"TRANSFORM {transformation}")
+    common.start_stream(cursor, "test")
     time.sleep(5)
 
     for topic in topics:
         producer.send(topic, SIMPLE_MSG).get(timeout=60)
 
     for topic in topics:
-        check_vertex_exists_with_topic_and_payload(
+        common.check_vertex_exists_with_topic_and_payload(
             cursor, topic, SIMPLE_MSG)
 
 
@@ -196,13 +46,13 @@ def test_separate_consumers(producer, topics, connection, transformation):
     for topic in topics:
         stream_name = "stream_" + topic
         stream_names.append(stream_name)
-        execute_and_fetch_all(cursor,
-                              f"CREATE STREAM {stream_name} "
-                              f"TOPICS {topic} "
-                              f"TRANSFORM {transformation}")
+        common.execute_and_fetch_all(cursor,
+                                     f"CREATE STREAM {stream_name} "
+                                     f"TOPICS {topic} "
+                                     f"TRANSFORM {transformation}")
 
     for stream_name in stream_names:
-        start_stream(cursor, stream_name)
+        common.start_stream(cursor, stream_name)
 
     time.sleep(5)
 
@@ -210,7 +60,7 @@ def test_separate_consumers(producer, topics, connection, transformation):
         producer.send(topic, SIMPLE_MSG).get(timeout=60)
 
     for topic in topics:
-        check_vertex_exists_with_topic_and_payload(
+        common.check_vertex_exists_with_topic_and_payload(
             cursor, topic, SIMPLE_MSG)
 
 
@@ -223,41 +73,42 @@ def test_start_from_last_committed_offset(producer, topics, connection):
     # restarting Memgraph during a single workload cannot be done currently.
     assert len(topics) > 0
     cursor = connection.cursor()
-    execute_and_fetch_all(cursor,
-                          "CREATE STREAM test "
-                          f"TOPICS {topics[0]} "
-                          "TRANSFORM transform.simple")
-    start_stream(cursor, "test")
+    common.execute_and_fetch_all(cursor,
+                                 "CREATE STREAM test "
+                                 f"TOPICS {topics[0]} "
+                                 "TRANSFORM transform.simple")
+    common.start_stream(cursor, "test")
     time.sleep(1)
 
     producer.send(topics[0], SIMPLE_MSG).get(timeout=60)
 
-    check_vertex_exists_with_topic_and_payload(
+    common.check_vertex_exists_with_topic_and_payload(
         cursor, topics[0], SIMPLE_MSG)
 
-    stop_stream(cursor, "test")
-    drop_stream(cursor, "test")
+    common.stop_stream(cursor, "test")
+    common.drop_stream(cursor, "test")
 
     messages = [b"second message", b"third message"]
     for message in messages:
         producer.send(topics[0], message).get(timeout=60)
 
     for message in messages:
-        vertices_with_msg = execute_and_fetch_all(cursor,
-                                                  "MATCH (n: MESSAGE {"
-                                                  f"payload: '{message.decode('utf-8')}'"
-                                                  "}) RETURN n")
+        vertices_with_msg = common.execute_and_fetch_all(
+            cursor,
+            "MATCH (n: MESSAGE {"
+            f"payload: '{message.decode('utf-8')}'"
+            "}) RETURN n")
 
         assert len(vertices_with_msg) == 0
 
-    execute_and_fetch_all(cursor,
-                          "CREATE STREAM test "
-                          f"TOPICS {topics[0]} "
-                          "TRANSFORM transform.simple")
-    start_stream(cursor, "test")
+    common.execute_and_fetch_all(cursor,
+                                 "CREATE STREAM test "
+                                 f"TOPICS {topics[0]} "
+                                 "TRANSFORM transform.simple")
+    common.start_stream(cursor, "test")
 
     for message in messages:
-        check_vertex_exists_with_topic_and_payload(
+        common.check_vertex_exists_with_topic_and_payload(
             cursor, topics[0], message)
 
 
@@ -265,16 +116,16 @@ def test_start_from_last_committed_offset(producer, topics, connection):
 def test_check_stream(producer, topics, connection, transformation):
     assert len(topics) > 0
     cursor = connection.cursor()
-    execute_and_fetch_all(cursor,
-                          "CREATE STREAM test "
-                          f"TOPICS {topics[0]} "
-                          f"TRANSFORM {transformation} "
-                          "BATCH_SIZE 1")
-    start_stream(cursor, "test")
+    common.execute_and_fetch_all(cursor,
+                                 "CREATE STREAM test "
+                                 f"TOPICS {topics[0]} "
+                                 f"TRANSFORM {transformation} "
+                                 "BATCH_SIZE 1")
+    common.start_stream(cursor, "test")
     time.sleep(1)
 
     producer.send(topics[0], SIMPLE_MSG).get(timeout=60)
-    stop_stream(cursor, "test")
+    common.stop_stream(cursor, "test")
 
     messages = [b"first message", b"second message", b"third message"]
     for message in messages:
@@ -283,7 +134,7 @@ def test_check_stream(producer, topics, connection, transformation):
     def check_check_stream(batch_limit):
         assert transformation == "transform.simple" \
             or transformation == "transform.with_parameters"
-        test_results = execute_and_fetch_all(
+        test_results = common.execute_and_fetch_all(
             cursor, f"CHECK STREAM test BATCH_LIMIT {batch_limit}")
         assert len(test_results) == batch_limit
 
@@ -308,42 +159,47 @@ def test_check_stream(producer, topics, connection, transformation):
     check_check_stream(1)
     check_check_stream(2)
     check_check_stream(3)
-    start_stream(cursor, "test")
+    common.start_stream(cursor, "test")
 
     for message in messages:
-        check_vertex_exists_with_topic_and_payload(
+        common.check_vertex_exists_with_topic_and_payload(
             cursor, topics[0], message)
 
 
 def test_show_streams(producer, topics, connection):
     assert len(topics) > 1
     cursor = connection.cursor()
-    execute_and_fetch_all(cursor,
-                          "CREATE STREAM default_values "
-                          f"TOPICS {topics[0]} "
-                          f"TRANSFORM transform.simple")
+    common.execute_and_fetch_all(cursor,
+                                 "CREATE STREAM default_values "
+                                 f"TOPICS {topics[0]} "
+                                 f"TRANSFORM transform.simple")
 
     consumer_group = "my_special_consumer_group"
     batch_interval = 42
     batch_size = 3
-    execute_and_fetch_all(cursor,
-                          "CREATE STREAM complex_values "
-                          f"TOPICS {','.join(topics)} "
-                          f"TRANSFORM transform.with_parameters "
-                          f"CONSUMER_GROUP {consumer_group} "
-                          f"BATCH_INTERVAL {batch_interval} "
-                          f"BATCH_SIZE {batch_size} ")
+    common.execute_and_fetch_all(cursor,
+                                 "CREATE STREAM complex_values "
+                                 f"TOPICS {','.join(topics)} "
+                                 f"TRANSFORM transform.with_parameters "
+                                 f"CONSUMER_GROUP {consumer_group} "
+                                 f"BATCH_INTERVAL {batch_interval} "
+                                 f"BATCH_SIZE {batch_size} ")
 
-    assert len(execute_and_fetch_all(cursor, "SHOW STREAMS")) == 2
+    assert len(common.execute_and_fetch_all(cursor, "SHOW STREAMS")) == 2
 
-    check_stream_info(cursor, "default_values", ("default_values", [
-                      topics[0]], "mg_consumer", None, None,
-        "transform.simple", False))
+    common.check_stream_info(cursor, "default_values", ("default_values", [
+        topics[0]], "mg_consumer", None, None,
+        "transform.simple", None, False))
 
-    check_stream_info(cursor, "complex_values", ("complex_values", topics,
-                      consumer_group, batch_interval, batch_size,
-                                                 "transform.with_parameters",
-                                                 False))
+    common.check_stream_info(cursor, "complex_values", (
+        "complex_values",
+        topics,
+        consumer_group,
+        batch_interval,
+        batch_size,
+        "transform.with_parameters",
+        None,
+        False))
 
 
 @pytest.mark.parametrize("operation", ["START", "STOP"])
@@ -360,10 +216,10 @@ def test_start_and_stop_during_check(producer, topics, connection, operation):
     assert len(topics) > 1
     assert operation == "START" or operation == "STOP"
     cursor = connection.cursor()
-    execute_and_fetch_all(cursor,
-                          "CREATE STREAM test_stream "
-                          f"TOPICS {topics[0]} "
-                          f"TRANSFORM transform.simple")
+    common.execute_and_fetch_all(cursor,
+                                 "CREATE STREAM test_stream "
+                                 f"TOPICS {topics[0]} "
+                                 f"TRANSFORM transform.simple")
 
     check_counter = Value('i', 0)
     check_result_len = Value('i', 0)
@@ -377,10 +233,11 @@ def test_start_and_stop_during_check(producer, topics, connection, operation):
     def call_check(counter, result_len):
         # This process will call the CHECK query and increment the counter
         # based on its progress and expected behavior
-        connection = connect()
+        connection = common.connect()
         cursor = connection.cursor()
         counter.value = CHECK_BEFORE_EXECUTE
-        result = execute_and_fetch_all(cursor, "CHECK STREAM test_stream")
+        result = common.execute_and_fetch_all(
+            cursor, "CHECK STREAM test_stream")
         result_len.value = len(result)
         counter.value = CHECK_AFTER_FETCHALL
         if len(result) > 0 and "payload: 'message'" in result[0][QUERY]:
@@ -397,11 +254,12 @@ def test_start_and_stop_during_check(producer, topics, connection, operation):
     def call_operation(counter):
         # This porcess will call the query with the specified operation and
         # increment the counter based on its progress and expected behavior
-        connection = connect()
+        connection = common.connect()
         cursor = connection.cursor()
         counter.value = OP_BEFORE_EXECUTE
         try:
-            execute_and_fetch_all(cursor, f"{operation} STREAM test_stream")
+            common.execute_and_fetch_all(
+                cursor, f"{operation} STREAM test_stream")
             counter.value = OP_AFTER_FETCHALL
         except mgclient.DatabaseError as e:
             if "Kafka consumer test_stream is already stopped" in str(e):
@@ -421,15 +279,19 @@ def test_start_and_stop_during_check(producer, topics, connection, operation):
 
         time.sleep(0.5)
 
-        assert timed_wait(lambda: check_counter.value == CHECK_BEFORE_EXECUTE)
-        assert timed_wait(lambda: get_is_running(cursor, "test_stream"))
+        assert common.timed_wait(
+            lambda: check_counter.value == CHECK_BEFORE_EXECUTE)
+        assert common.timed_wait(
+            lambda: common.get_is_running(cursor, "test_stream"))
         assert check_counter.value == CHECK_BEFORE_EXECUTE, "SHOW STREAMS " \
             "was blocked until the end of CHECK STREAM"
         operation_proc.start()
-        assert timed_wait(lambda: operation_counter.value == OP_BEFORE_EXECUTE)
+        assert common.timed_wait(
+            lambda: operation_counter.value == OP_BEFORE_EXECUTE)
 
         producer.send(topics[0], SIMPLE_MSG).get(timeout=60)
-        assert timed_wait(lambda: check_counter.value > CHECK_AFTER_FETCHALL)
+        assert common.timed_wait(
+            lambda: check_counter.value > CHECK_AFTER_FETCHALL)
         assert check_counter.value == CHECK_CORRECT_RESULT
         assert check_result_len.value == 1
         check_stream_proc.join()
@@ -437,10 +299,10 @@ def test_start_and_stop_during_check(producer, topics, connection, operation):
         operation_proc.join()
         if operation == "START":
             assert operation_counter.value == OP_AFTER_FETCHALL
-            assert get_is_running(cursor, "test_stream")
+            assert common.get_is_running(cursor, "test_stream")
         else:
             assert operation_counter.value == OP_ALREADY_STOPPED_EXCEPTION
-            assert not get_is_running(cursor, "test_stream")
+            assert not common.get_is_running(cursor, "test_stream")
 
     finally:
         # to make sure CHECK STREAM finishes
@@ -455,42 +317,64 @@ def test_check_already_started_stream(topics, connection):
     assert len(topics) > 0
     cursor = connection.cursor()
 
-    execute_and_fetch_all(cursor,
-                          "CREATE STREAM started_stream "
-                          f"TOPICS {topics[0]} "
-                          f"TRANSFORM transform.simple")
-    start_stream(cursor, "started_stream")
+    common.execute_and_fetch_all(cursor,
+                                 "CREATE STREAM started_stream "
+                                 f"TOPICS {topics[0]} "
+                                 f"TRANSFORM transform.simple")
+    common.start_stream(cursor, "started_stream")
 
     with pytest.raises(mgclient.DatabaseError):
-        execute_and_fetch_all(cursor, "CHECK STREAM started_stream")
+        common.execute_and_fetch_all(cursor, "CHECK STREAM started_stream")
 
 
 def test_start_checked_stream_after_timeout(topics, connection):
     cursor = connection.cursor()
-    execute_and_fetch_all(cursor,
-                          "CREATE STREAM test_stream "
-                          f"TOPICS {topics[0]} "
-                          f"TRANSFORM transform.simple")
+    common.execute_and_fetch_all(cursor,
+                                 "CREATE STREAM test_stream "
+                                 f"TOPICS {topics[0]} "
+                                 f"TRANSFORM transform.simple")
 
     timeout_ms = 2000
 
     def call_check():
-        execute_and_fetch_all(
-            connect().cursor(),
+        common.execute_and_fetch_all(
+            common.connect().cursor(),
             f"CHECK STREAM test_stream TIMEOUT {timeout_ms}")
 
     check_stream_proc = Process(target=call_check, daemon=True)
 
     start = time.time()
     check_stream_proc.start()
-    assert timed_wait(lambda: get_is_running(cursor, "test_stream"))
-    start_stream(cursor, "test_stream")
+    assert common.timed_wait(
+        lambda: common.get_is_running(cursor, "test_stream"))
+    common.start_stream(cursor, "test_stream")
     end = time.time()
 
     assert (end - start) < 1.3 * \
         timeout_ms, "The START STREAM was blocked too long"
-    assert get_is_running(cursor, "test_stream")
-    stop_stream(cursor, "test_stream")
+    assert common.get_is_running(cursor, "test_stream")
+    common.stop_stream(cursor, "test_stream")
+
+
+def test_restart_after_error(producer, topics, connection):
+    cursor = connection.cursor()
+    common.execute_and_fetch_all(cursor,
+                                 "CREATE STREAM test_stream "
+                                 f"TOPICS {topics[0]} "
+                                 f"TRANSFORM transform.query")
+
+    common.start_stream(cursor, "test_stream")
+    time.sleep(1)
+
+    producer.send(topics[0], SIMPLE_MSG).get(timeout=60)
+    assert common.timed_wait(
+        lambda: not common.get_is_running(cursor, "test_stream"))
+
+    common.start_stream(cursor, "test_stream")
+    time.sleep(1)
+    producer.send(topics[0], b'CREATE (n:VERTEX { id : 42 })')
+    assert common.check_one_result_row(
+        cursor, "MATCH (n:VERTEX { id : 42 }) RETURN n")
 
 
 if __name__ == "__main__":

tests/e2e/streams/transformations/transform.py

@@ -35,3 +35,17 @@ def with_parameters(context: mgp.TransCtx,
                         "topic": message.topic_name()}))
 
     return result_queries
+
+
+@mgp.transformation
+def query(messages: mgp.Messages
+          ) -> mgp.Record(query=str, parameters=mgp.Nullable[mgp.Map]):
+    result_queries = []
+
+    for i in range(0, messages.total_messages()):
+        message = messages.message_at(i)
+        payload_as_str = message.payload().decode("utf-8")
+        result_queries.append(mgp.Record(
+            query=payload_as_str, parameters=None))
+
+    return result_queries

tests/e2e/streams/workloads.yaml

@@ -10,5 +10,10 @@ workloads:
   - name: "Streams start, stop and show"
     binary: "tests/e2e/streams/streams_test_runner.sh"
     proc: "tests/e2e/streams/transformations/"
-    args: []
+    args: ["streams_tests.py"]
+    <<: *template_cluster
+  - name: "Streams with users"
+    binary: "tests/e2e/streams/streams_test_runner.sh"
+    proc: "tests/e2e/streams/transformations/"
+    args: ["streams_owner_tests.py"]
     <<: *template_cluster

tests/e2e/triggers/CMakeLists.txt

@@ -9,3 +9,6 @@ target_link_libraries(memgraph__e2e__triggers__on_update memgraph__e2e__triggers
 
 add_executable(memgraph__e2e__triggers__on_delete on_delete_triggers.cpp)
 target_link_libraries(memgraph__e2e__triggers__on_delete memgraph__e2e__triggers_common)
+
+add_executable(memgraph__e2e__triggers__privileges privilige_check.cpp)
+target_link_libraries(memgraph__e2e__triggers__privileges memgraph__e2e__triggers_common)

tests/e2e/triggers/common.cpp

@@ -1,7 +1,9 @@
 #include "common.hpp"
 
+#include <chrono>
 #include <cstdint>
 #include <optional>
+#include <thread>
 
 #include <fmt/format.h>
 #include <gflags/gflags.h>
@@ -10,13 +12,17 @@
 
 DEFINE_uint64(bolt_port, 7687, "Bolt port");
 
-std::unique_ptr<mg::Client> Connect() {
-  auto client =
-      mg::Client::Connect({.host = "127.0.0.1", .port = static_cast<uint16_t>(FLAGS_bolt_port), .use_ssl = false});
+std::unique_ptr<mg::Client> ConnectWithUser(const std::string_view username) {
+  auto client = mg::Client::Connect({.host = "127.0.0.1",
+                                     .port = static_cast<uint16_t>(FLAGS_bolt_port),
+                                     .username = std::string{username},
+                                     .use_ssl = false});
   MG_ASSERT(client, "Failed to connect!");
   return client;
 }
 
+std::unique_ptr<mg::Client> Connect() { return ConnectWithUser(""); }
+
 void CreateVertex(mg::Client &client, int vertex_id) {
   mg::Map parameters{
       {"id", mg::Value{vertex_id}},
@@ -49,10 +55,12 @@ int GetNumberOfAllVertices(mg::Client &client) {
 }
 
 void WaitForNumberOfAllVertices(mg::Client &client, int number_of_vertices) {
+  using namespace std::chrono_literals;
   utils::Timer timer{};
   while ((timer.Elapsed().count() <= 0.5) && GetNumberOfAllVertices(client) != number_of_vertices) {
   }
   CheckNumberOfAllVertices(client, number_of_vertices);
+  std::this_thread::sleep_for(100ms);
 }
 
 void CheckNumberOfAllVertices(mg::Client &client, int expected_number_of_vertices) {

tests/e2e/triggers/common.hpp

@@ -11,6 +11,7 @@ constexpr std::string_view kVertexLabel{"VERTEX"};
 constexpr std::string_view kEdgeLabel{"EDGE"};
 
 std::unique_ptr<mg::Client> Connect();
+std::unique_ptr<mg::Client> ConnectWithUser(const std::string_view username);
 void CreateVertex(mg::Client &client, int vertex_id);
 void CreateEdge(mg::Client &client, int from_vertex, int to_vertex, int edge_id);
 

tests/e2e/triggers/privilige_check.cpp

@@ -0,0 +1,161 @@
+#include <string>
+#include <string_view>
+
+#include <gflags/gflags.h>
+#include <spdlog/fmt/bundled/core.h>
+#include <mgclient.hpp>
+#include "common.hpp"
+#include "utils/logging.hpp"
+
+constexpr std::string_view kTriggerPrefix{"CreatedVerticesTrigger"};
+
+int main(int argc, char **argv) {
+  gflags::SetUsageMessage("Memgraph E2E Triggers privilege check");
+  gflags::ParseCommandLineFlags(&argc, &argv, true);
+  logging::RedirectToStderr();
+
+  constexpr int kVertexId{42};
+  constexpr std::string_view kUserlessLabel{"USERLESS"};
+  constexpr std::string_view kAdminUser{"ADMIN"};
+  constexpr std::string_view kUserWithCreate{"USER_WITH_CREATE"};
+  constexpr std::string_view kUserWithoutCreate{"USER_WITHOUT_CREATE"};
+
+  mg::Client::Init();
+
+  auto userless_client = Connect();
+
+  const auto get_number_of_triggers = [&userless_client] {
+    userless_client->Execute("SHOW TRIGGERS");
+    auto result = userless_client->FetchAll();
+    MG_ASSERT(result.has_value());
+    return result->size();
+  };
+
+  auto create_trigger = [&get_number_of_triggers](mg::Client &client, const std::string_view vertexLabel,
+                                                  bool should_succeed = true) {
+    const auto number_of_triggers_before = get_number_of_triggers();
+    client.Execute(
+        fmt::format("CREATE TRIGGER {}{} ON () CREATE "
+                    "AFTER COMMIT "
+                    "EXECUTE "
+                    "UNWIND createdVertices as createdVertex "
+                    "CREATE (n: {} {{ id: createdVertex.id }})",
+                    kTriggerPrefix, vertexLabel, vertexLabel));
+    client.DiscardAll();
+    const auto number_of_triggers_after = get_number_of_triggers();
+    if (should_succeed) {
+      MG_ASSERT(number_of_triggers_after == number_of_triggers_before + 1);
+    } else {
+      MG_ASSERT(number_of_triggers_after == number_of_triggers_before);
+    }
+  };
+
+  auto delete_vertices = [&userless_client] {
+    userless_client->Execute("MATCH (n) DETACH DELETE n;");
+    userless_client->DiscardAll();
+    CheckNumberOfAllVertices(*userless_client, 0);
+  };
+
+  auto create_user = [&userless_client](const std::string_view username) {
+    userless_client->Execute(fmt::format("CREATE USER {};", username));
+    userless_client->DiscardAll();
+    userless_client->Execute(fmt::format("GRANT TRIGGER TO {};", username));
+    userless_client->DiscardAll();
+  };
+
+  auto drop_user = [&userless_client](const std::string_view username) {
+    userless_client->Execute(fmt::format("DROP USER {};", username));
+    userless_client->DiscardAll();
+  };
+
+  auto drop_trigger_of_user = [&userless_client](const std::string_view username) {
+    userless_client->Execute(fmt::format("DROP TRIGGER {}{};", kTriggerPrefix, username));
+    userless_client->DiscardAll();
+  };
+
+  // Single trigger created without user, there is no existing users
+  create_trigger(*userless_client, kUserlessLabel);
+  CreateVertex(*userless_client, kVertexId);
+
+  WaitForNumberOfAllVertices(*userless_client, 2);
+  CheckVertexExists(*userless_client, kVertexLabel, kVertexId);
+  CheckVertexExists(*userless_client, kUserlessLabel, kVertexId);
+
+  delete_vertices();
+
+  // Single trigger created without user, there is an existing user
+  // The trigger fails because there is no owner
+  create_user(kAdminUser);
+  CreateVertex(*userless_client, kVertexId);
+
+  CheckVertexExists(*userless_client, kVertexLabel, kVertexId);
+  CheckNumberOfAllVertices(*userless_client, 1);
+
+  delete_vertices();
+
+  // Three triggers: without an owner, an owner with CREATE privilege, an owner without CREATE privilege; there are
+  // existing users
+  // Only the trigger which owner has CREATE privilege will succeed
+  create_user(kUserWithCreate);
+  userless_client->Execute(fmt::format("GRANT CREATE TO {};", kUserWithCreate));
+  userless_client->DiscardAll();
+  create_user(kUserWithoutCreate);
+  auto client_with_create = ConnectWithUser(kUserWithCreate);
+  auto client_without_create = ConnectWithUser(kUserWithoutCreate);
+  create_trigger(*client_with_create, kUserWithCreate);
+  create_trigger(*client_without_create, kUserWithoutCreate, false);
+
+  // Grant CREATE to be able to create the trigger than revoke it
+  userless_client->Execute(fmt::format("GRANT CREATE TO {};", kUserWithoutCreate));
+  userless_client->DiscardAll();
+  create_trigger(*client_without_create, kUserWithoutCreate);
+  userless_client->Execute(fmt::format("REVOKE CREATE FROM {};", kUserWithoutCreate));
+  userless_client->DiscardAll();
+
+  CreateVertex(*userless_client, kVertexId);
+
+  WaitForNumberOfAllVertices(*userless_client, 2);
+  CheckVertexExists(*userless_client, kVertexLabel, kVertexId);
+  CheckVertexExists(*userless_client, kUserWithCreate, kVertexId);
+
+  delete_vertices();
+
+  // Three triggers: without an owner, an owner with CREATE privilege, an owner without CREATE privilege; there is no
+  // existing user
+  // All triggers will succeed, as there is no authorization is done when there are no users
+  drop_user(kAdminUser);
+  drop_user(kUserWithCreate);
+  drop_user(kUserWithoutCreate);
+  CreateVertex(*userless_client, kVertexId);
+
+  WaitForNumberOfAllVertices(*userless_client, 4);
+  CheckVertexExists(*userless_client, kVertexLabel, kVertexId);
+  CheckVertexExists(*userless_client, kUserlessLabel, kVertexId);
+  CheckVertexExists(*userless_client, kUserWithCreate, kVertexId);
+  CheckVertexExists(*userless_client, kUserWithoutCreate, kVertexId);
+
+  delete_vertices();
+  drop_trigger_of_user(kUserlessLabel);
+  drop_trigger_of_user(kUserWithCreate);
+  drop_trigger_of_user(kUserWithoutCreate);
+
+  // The BEFORE COMMIT trigger without proper privileges make the transaction fail
+  create_user(kUserWithoutCreate);
+  userless_client->Execute(fmt::format("GRANT CREATE TO {};", kUserWithoutCreate));
+  userless_client->DiscardAll();
+  client_without_create->Execute(
+      fmt::format("CREATE TRIGGER {}{} ON () CREATE "
+                  "BEFORE COMMIT "
+                  "EXECUTE "
+                  "UNWIND createdVertices as createdVertex "
+                  "CREATE (n: {} {{ id: createdVertex.id }})",
+                  kTriggerPrefix, kUserWithoutCreate, kUserWithoutCreate));
+  client_without_create->DiscardAll();
+  userless_client->Execute(fmt::format("REVOKE CREATE FROM {};", kUserWithoutCreate));
+  userless_client->DiscardAll();
+
+  CreateVertex(*userless_client, kVertexId);
+  CheckNumberOfAllVertices(*userless_client, 0);
+
+  return 0;
+}

tests/e2e/triggers/workloads.yaml

@@ -20,5 +20,9 @@ workloads:
     binary: "tests/e2e/triggers/memgraph__e2e__triggers__on_delete"
     args: ["--bolt-port", *bolt_port]
     <<: *template_cluster
+  - name: "Triggers privilege check"
+    binary: "tests/e2e/triggers/memgraph__e2e__triggers__privileges"
+    args: ["--bolt-port", *bolt_port]
+    <<: *template_cluster