diff --git a/openserverless/actionexecutor.py b/openserverless/actionexecutor.py index 1a69a94..e4fdcca 100644 --- a/openserverless/actionexecutor.py +++ b/openserverless/actionexecutor.py @@ -22,6 +22,9 @@ import requests as req from datetime import datetime +# Bound tenant-controlled input before passing it to the cron parser. +MAX_CRON_EXPRESSION_LENGTH = 4096 + def check(f, what, res): if f: logging.info(f"OK: {what}") @@ -244,8 +247,11 @@ def get_autoexec(actionAnnotations): # no_execution if the action should not be triggered or the cron expression is not a valid one # def should_trigger(actionNamespace, actionName, actionCronExpression, currentDate, executionInterval): - if not cn.croniter.is_valid(actionCronExpression): - logging.warn(f"action {actionNamespace}/{actionName} cron expression {actionCronExpression} is not valid. Skipping execution") + if (not isinstance(actionCronExpression, str) + or not 0 < len(actionCronExpression) <= MAX_CRON_EXPRESSION_LENGTH + or not cn.croniter.is_valid(actionCronExpression)): + logging.warning("action %s/%s has an invalid cron expression. Skipping execution", + actionNamespace, actionName) return "no_execution" if not action_should_trigger(currentDate, executionInterval, actionCronExpression): @@ -389,8 +395,11 @@ def start(): if(len(actions) > 0): subjects = get_subjects(db, config['couchdb.controller.user'],config['couchdb.controller.password']) for action in actions: - handle_action(baseurl, currentDate, interval, action, subjects) + try: + handle_action(baseurl, currentDate, interval, action, subjects) + except Exception: + logging.exception("Failed to process scheduled action; continuing with remaining actions") else: logging.info('No cron aware action extracted. Exiting....') else: - logging.warn("CouchDB it is not available. Exiting....") \ No newline at end of file + logging.warn("CouchDB it is not available. Exiting....") diff --git a/tests/test_actionexecutor_cron_security.py b/tests/test_actionexecutor_cron_security.py new file mode 100644 index 0000000..94721a1 --- /dev/null +++ b/tests/test_actionexecutor_cron_security.py @@ -0,0 +1,112 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +import json +import unittest +from unittest.mock import Mock, patch + +import openserverless.actionexecutor as executor + + +def action(namespace, name, annotations): + return { + "name": name, + "entityType": "action", + "namespace": namespace, + "parameters": [], + "annotations": annotations, + } + + +class CronTenantIsolationTest(unittest.TestCase): + def test_invalid_cron_values_are_skipped_before_parsing(self): + for value in (None, False, 42, [], {}, "", "x" * 4097): + with self.subTest(value=value): + with patch.object(executor.cn.croniter, "is_valid") as is_valid, self.assertLogs(level="WARNING"): + result = executor.should_trigger( + "tenant-a", "broken", value, executor.datetime.now(), 60 + ) + self.assertEqual(result, "no_execution") + is_valid.assert_not_called() + + def test_malformed_actions_do_not_stop_other_tenants(self): + actions = [ + action("tenant-a", "null-cron", [{"key": "cron", "value": None}]), + action("tenant-a", "missing-value", [{"key": "cron"}]), + action("tenant-b", "healthy", [{"key": "cron", "value": "* * * * *"}]), + action("tenant-b", "one-shot", [{"key": "autoexec", "value": True}]), + ] + subjects = [ + {"name": "tenant-a", "uuid": "a", "key": "a-secret"}, + {"name": "tenant-b", "uuid": "b", "key": "b-secret"}, + ] + config = { + "scheduler.schedule": "* * * * *", + "controller.protocol": "http", + "controller.host": "controller", + "controller.port": "3233", + "couchdb.controller.user": "controller", + "couchdb.controller.password": "secret", + } + db = Mock() + db.wait_db_ready.return_value = True + batches = { + "whisks": [ + {"docs": actions, "bookmark": "next-actions"}, + {"docs": [], "bookmark": ""}, + ], + "subjects": [ + {"docs": [{"namespaces": subjects}], "bookmark": "next-subjects"}, + {"docs": [], "bookmark": ""}, + ], + } + + def find_doc(database, selector, username, password): + self.assertEqual((username, password), ("controller", "secret")) + self.assertIn("selector", json.loads(selector)) + return batches[database].pop(0) + + db.find_doc.side_effect = find_doc + + with ( + patch.dict(executor.os.environ, {"SCHEDULER_CONFIG": json.dumps(config)}), + patch.object(executor.cu, "CouchDB", return_value=db), + patch.object(executor.req, "post", return_value=Mock(status_code=202, text="")) as invoke, + patch.object(executor.req, "put", return_value=Mock(status_code=200, text="")) as unschedule, + self.assertLogs(level="WARNING") as logs, + ): + executor.start() + + self.assertTrue(any("Failed to process scheduled action" in line for line in logs.output)) + self.assertEqual(invoke.call_count, 2) + self.assertEqual( + [call.args[0] for call in invoke.call_args_list], + [ + "http://controller:3233/api/v1/namespaces/tenant-b/actions/healthy?blocking=false&result=false", + "http://controller:3233/api/v1/namespaces/tenant-b/actions/one-shot?blocking=false&result=false", + ], + ) + self.assertEqual( + [call.kwargs["auth"] for call in invoke.call_args_list], + [("b", "b-secret"), ("b", "b-secret")], + ) + unschedule.assert_called_once() + self.assertEqual(unschedule.call_args.kwargs["auth"], ("b", "b-secret")) + + +if __name__ == "__main__": + unittest.main()