Merge pull request #36 from timohaas/bugfix

Prevent users from creating databases with owner `postgres`

CHANGES.rst

@@ -2,6 +2,11 @@
 Changelog
 =========
 
+v2.1.2
+======
+
+- Blacklist postgres as username
+
 v2.1.1
 ======
 

postgraas_server/management_resources.py

@@ -137,6 +137,12 @@ def post(self):
         if not args['db_pwd']:
             abort(400, msg='The password may not be empty.')
 
+        if args['db_username'] == "postgres" or  args['db_username'].startswith("postgres@"):
+            abort(
+                422,
+                msg="username {} is backlisted".format(args['db_username'])
+            )
+
         if DBInstance.query.filter_by(postgraas_instance_name=args['postgraas_instance_name']
                                       ).first():
             abort(

tests/test_integration/test_postgraas_api.py

@@ -176,6 +176,51 @@ def test_create_postgres_instance_api_with_fully_qualified_user(self):
         assert created_db["db_name"] == 'test_create_postgres_instance'
         self.delete_instance_by_name(db_credentials, self.app_client)
 
+    def test_create_postgres_instance_api_with_postgres_as_user(self):
+        db_credentials = {
+            "postgraas_instance_name": "tests_postgraas_test_create_postgres_instance_api",
+            "db_name": "test_create_postgres_instance",
+            "db_username": "postgres",
+            "db_pwd": "secret"
+        }
+        self.delete_instance_by_name(db_credentials, self.app_client)
+        headers = {'Content-Type': 'application/json'}
+        result = self.app_client.post(
+            '/api/v2/postgraas_instances', headers=headers, data=json.dumps(db_credentials)
+        )
+        assert result.status_code == 422
+        self.delete_instance_by_name(db_credentials, self.app_client)
+
+    def test_create_postgres_instance_api_with_postgres_at_example_com_as_user(self):
+        db_credentials = {
+            "postgraas_instance_name": "tests_postgraas_test_create_postgres_instance_api",
+            "db_name": "test_create_postgres_instance",
+            "db_username": "postgres@example.com",
+            "db_pwd": "secret"
+        }
+        self.delete_instance_by_name(db_credentials, self.app_client)
+        headers = {'Content-Type': 'application/json'}
+        result = self.app_client.post(
+            '/api/v2/postgraas_instances', headers=headers, data=json.dumps(db_credentials)
+        )
+        assert result.status_code == 422
+        self.delete_instance_by_name(db_credentials, self.app_client)
+
+    def test_create_postgres_instance_api_with_postgres_at_localhost_as_user(self):
+        db_credentials = {
+            "postgraas_instance_name": "tests_postgraas_test_create_postgres_instance_api",
+            "db_name": "test_create_postgres_instance",
+            "db_username": "postgres@localhost",
+            "db_pwd": "secret"
+        }
+        self.delete_instance_by_name(db_credentials, self.app_client)
+        headers = {'Content-Type': 'application/json'}
+        result = self.app_client.post(
+            '/api/v2/postgraas_instances', headers=headers, data=json.dumps(db_credentials)
+        )
+        assert result.status_code == 422
+        self.delete_instance_by_name(db_credentials, self.app_client)
+
     def test_create_docker_fails(self):
         db_credentials = {
             "postgraas_instance_name": "tests_postgraas_test_create_postgres_instance_api",