Start implementing user, quotas and roles

docs/developer/rest-api.rst

@@ -90,7 +90,7 @@ Will return a JSON document like this::
        "error_message" : null,
        "time_start" : 1473337160.16264,
        "id" : 25158,
-       "user_id" : "venzano",
+       "user_id" : 1,
        "time_end" : null,
        "name" : "boinc-loader",
        "services" : [
@@ -160,7 +160,7 @@ Will return a JSON document like this::
        "25086" : {
           "time_start" : 1473337123.30892,
           "status" : "running",
-          "user_id" : "venzano",
+          "user_id" : 1,
     [..]
 
 It is a map with the execution IDs as keys and the full execution details as values.

zoe_api/api_endpoint.py

@@ -28,8 +28,6 @@ from zoe_lib.config import get_conf
 
 log = logging.getLogger(__name__)
 
-GUEST_QUOTA_MAX_EXECUTIONS = 1
-
 
 class APIEndpoint:
     """
@@ -42,27 +40,27 @@ class APIEndpoint:
         self.master = master_api
         self.sql = sql_manager
 
-    def execution_by_id(self, uid, role, execution_id) -> zoe_lib.state.Execution:
+    def execution_by_id(self, uid: zoe_lib.state.User, role: zoe_lib.state.Role, execution_id: int) -> zoe_lib.state.Execution:
         """Lookup an execution by its ID."""
         e = self.sql.executions.select(id=execution_id, only_one=True)
         if e is None:
             raise zoe_api.exceptions.ZoeNotFoundException('No such execution')
         assert isinstance(e, zoe_lib.state.Execution)
-        if e.user_id != uid and role != 'admin':
+        if e.user_id != uid.id and not role.can_operate_others:
             raise zoe_api.exceptions.ZoeAuthException()
         return e
 
-    def execution_list(self, uid: str, role: str, **filters: Mapping[str, str]):
+    def execution_list(self, uid: zoe_lib.state.User, role: zoe_lib.state.Role, **filters: Mapping[str, str]):
         """Generate a optionally filtered list of executions."""
-        if role != 'admin':
-            filters['user_id'] = uid
+        if not role.can_operate_others:
+            filters['user_id'] = uid.id
         execs = self.sql.executions.select(**filters)
         return execs
 
-    def execution_count(self, uid: str, role: str, **filters: Mapping[str, str]):
+    def execution_count(self, uid: zoe_lib.state.User, role: zoe_lib.state.Role, **filters: Mapping[str, str]):
         """Count the number of executions optionally filtered."""
-        if role != 'admin':
-            filters['user_id'] = uid
+        if not role.can_operate_others:
+            filters['user_id'] = uid.id
         execs = self.sql.executions.count(**filters)
         return execs
 
@@ -73,38 +71,44 @@ class APIEndpoint:
         except zoe_lib.exceptions.InvalidApplicationDescription as e:
             raise zoe_api.exceptions.ZoeException('Invalid application description: ' + e.message)
 
-    def execution_start(self, uid, role, exec_name, application_description):  # pylint: disable=unused-argument
+    def _check_quota(self, user: zoe_lib.state.User, application_description):  # pylint: disable=unused-argument
+        """Check quota for given user and execution."""
+        quota = self.sql.quota.select(only_one=True, **{'id': user.quota_id})
+
+        running_execs = self.sql.executions.select(**{'status': 'running', 'user_id': user.id})
+        running_execs += self.sql.executions.select(**{'status': 'starting', 'user_id': user.id})
+        running_execs += self.sql.executions.select(**{'status': 'scheduled', 'user_id': user.id})
+        running_execs += self.sql.executions.select(**{'status': 'image download', 'user_id': user.id})
+        running_execs += self.sql.executions.select(**{'status': 'submitted', 'user_id': user.id})
+        if len(running_execs) >= quota.concurrent_executions:
+            raise zoe_api.exceptions.ZoeException('You cannot run more than {} executions at a time, quota exceeded.'.format(quota.concurrent_executions))
+
+        # TODO: implement core and memory quotas
+
+    def execution_start(self, uid: zoe_lib.state.User, role: zoe_lib.state.Role, exec_name, application_description):  # pylint: disable=unused-argument
         """Start an execution."""
         try:
             zoe_lib.applications.app_validate(application_description)
         except zoe_lib.exceptions.InvalidApplicationDescription as e:
             raise zoe_api.exceptions.ZoeException('Invalid application description: ' + e.message)
 
-        # quota check
-        if role == "guest":
-            running_execs = self.execution_list(uid, role, **{'status': 'running'})
-            running_execs += self.execution_list(uid, role, **{'status': 'starting'})
-            running_execs += self.execution_list(uid, role, **{'status': 'scheduled'})
-            running_execs += self.execution_list(uid, role, **{'status': 'image download'})
-            running_execs += self.execution_list(uid, role, **{'status': 'submitted'})
-            if len(running_execs) >= GUEST_QUOTA_MAX_EXECUTIONS:
-                raise zoe_api.exceptions.ZoeException('Guest users cannot run more than one execution at a time, quota exceeded.')
-
-        new_id = self.sql.executions.insert(exec_name, uid, application_description)
+        self._check_quota(uid, application_description)
+
+        new_id = self.sql.executions.insert(exec_name, uid.id, application_description)
         success, message = self.master.execution_start(new_id)
         if not success:
             raise zoe_api.exceptions.ZoeException('The Zoe master is unavailable, execution will be submitted automatically when the master is back up ({}).'.format(message))
 
         return new_id
 
-    def execution_terminate(self, uid, role, exec_id):
+    def execution_terminate(self, uid: zoe_lib.state.User, role: zoe_lib.state.Role, exec_id: int):
         """Terminate an execution."""
         e = self.sql.executions.select(id=exec_id, only_one=True)
         assert isinstance(e, zoe_lib.state.Execution)
         if e is None:
             raise zoe_api.exceptions.ZoeNotFoundException('No such execution')
 
-        if e.user_id != uid and role != 'admin':
+        if e.user_id != uid.id and not role.can_operate_others:
             raise zoe_api.exceptions.ZoeAuthException()
 
         if e.is_active:
@@ -112,9 +116,9 @@ class APIEndpoint:
         else:
             raise zoe_api.exceptions.ZoeException('Execution is not running')
 
-    def execution_delete(self, uid, role, exec_id):
+    def execution_delete(self, uid: zoe_lib.state.User, role: zoe_lib.state.Role, exec_id: int):
         """Delete an execution."""
-        if role != "admin":
+        if not role.can_delete_executions:
             raise zoe_api.exceptions.ZoeAuthException()
 
         e = self.sql.executions.select(id=exec_id, only_one=True)
@@ -122,7 +126,7 @@ class APIEndpoint:
         if e is None:
             raise zoe_api.exceptions.ZoeNotFoundException('No such execution')
 
-        if e.user_id != uid and role != 'admin':
+        if e.user_id != uid.id and not role.can_operate_others:
             raise zoe_api.exceptions.ZoeAuthException()
 
         if e.is_active:
@@ -135,29 +139,29 @@ class APIEndpoint:
         else:
             raise zoe_api.exceptions.ZoeException(message)
 
-    def service_by_id(self, uid, role, service_id) -> zoe_lib.state.Service:
+    def service_by_id(self, uid: zoe_lib.state.User, role: zoe_lib.state.Role, service_id: int) -> zoe_lib.state.Service:
         """Lookup a service by its ID."""
         service = self.sql.services.select(id=service_id, only_one=True)
         if service is None:
             raise zoe_api.exceptions.ZoeNotFoundException('No such execution')
-        if service.user_id != uid and role != 'admin':
+        if service.user_id != uid.id and not role.can_operate_others:
             raise zoe_api.exceptions.ZoeAuthException()
         return service
 
-    def service_list(self, uid, role, **filters):
+    def service_list(self, uid: zoe_lib.state.User, role: zoe_lib.state.Role, **filters):
         """Generate a optionally filtered list of services."""
-        services = self.sql.services.select(**filters)
-        ret = [s for s in services if s.user_id == uid or role == 'admin']
-        return ret
+        if not role.can_operate_others:
+            filters['user_id'] = uid.id
+        return self.sql.services.select(**filters)
 
-    def service_logs(self, uid, role, service_id):
+    def service_logs(self, uid: zoe_lib.state.User, role: zoe_lib.state.Role, service_id):
         """Retrieve the logs for the given service.
         If stream is True, a file object is returned, otherwise the log contents as a str object.
         """
         service = self.sql.services.select(id=service_id, only_one=True)
         if service is None:
             raise zoe_api.exceptions.ZoeNotFoundException('No such service')
-        if service.user_id != uid and role != 'admin':
+        if service.user_id != uid.id and not role.can_operate_others:
             raise zoe_api.exceptions.ZoeAuthException()
 
         path = os.path.join(get_conf().service_logs_base_path, get_conf().deployment_name, str(service.execution_id), service.name + '.txt')
@@ -177,7 +181,7 @@ class APIEndpoint:
         else:
             raise zoe_api.exceptions.ZoeException(message=message)
 
-    def execution_endpoints(self, uid: str, role: str, execution: zoe_lib.state.Execution):
+    def execution_endpoints(self, uid: zoe_lib.state.User, role: zoe_lib.state.Role, execution: zoe_lib.state.Execution):
         """Return a list of the services and public endpoints available for a certain execution."""
         services_info = []
         endpoints = []

zoe_lib/state/__init__.py

@@ -20,3 +20,6 @@ from zoe_lib.state.execution import Execution
 from zoe_lib.state.sql_manager import SQLManager
 from zoe_lib.state.service import Service, VolumeDescription, VolumeDescriptionHostPath
 from zoe_lib.state.port import Port
+from zoe_lib.state.user import User
+from zoe_lib.state.role import Role
+from zoe_lib.state.quota import Quota

zoe_lib/state/execution.py

@@ -217,6 +217,11 @@ class Execution(BaseRecord):
             return 0
         return functools.reduce(lambda x, y: x + y, [s.resource_reservation for s in self.services])
 
+    @property
+    def owner(self):
+        """Returns the full user object that owns this execution."""
+        return self.sql_manager.user.select(only_one=True, **{'id': self.user_id})
+
     def __repr__(self):
         return str(self.id)
 
@@ -231,7 +236,7 @@ class ExecutionTable(BaseTable):
         self.cursor.execute('''CREATE TABLE execution (
             id SERIAL PRIMARY KEY,
             name TEXT NOT NULL,
-            user_id TEXT NOT NULL,
+            user_id INT REFERENCES "user",
             description JSON NOT NULL,
             status TEXT NOT NULL,
             size NUMERIC NOT NULL,

zoe_lib/state/quota.py

+# Copyright (c) 2017, Daniele Venzano
+#
+# Licensed 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.
+
+"""Interface to PostgresQL for Zoe state."""
+
+import logging
+
+from zoe_lib.state.base import BaseTable, BaseRecord
+
+log = logging.getLogger(__name__)
+
+
+class Quota(BaseRecord):
+    """A quota object describes limits imposed to users on resource usage."""
+
+    def __init__(self, d, sql_manager):
+        super().__init__(d, sql_manager)
+
+        self.name = d['name']
+        self.concurrent_executions = d['concurrent_executions']
+        self.memory = d['memory']
+        self.cores = d['cores']
+
+    def serialize(self):
+        """Generates a dictionary that can be serialized in JSON."""
+        return {
+            'id': self.id,
+            'name': self.name,
+            'concurrent_executions': self.concurrent_executions,
+            'cores': self.cores,
+            'memory': self.memory
+        }
+
+    def set_concurrent_executions(self, value):
+        """Setter for concurrent execution limit."""
+        self.concurrent_executions = value
+        self.sql_manager.quota_update(self.id, concurrent_executions=value)
+
+    def set_memory(self, value):
+        """Setter for memory limit."""
+        self.memory = value
+        self.sql_manager.quota_update(self.id, memory=value)
+
+    def set_cores(self, value):
+        """Setter for cores limit."""
+        self.cores = value
+        self.sql_manager.quota_update(self.id, cores=value)
+
+
+class QuotaTable(BaseTable):
+    """Abstraction for the quota table in the database."""
+    def __init__(self, sql_manager):
+        super().__init__(sql_manager, "quota")
+
+    def create(self):
+        """Create the quota table."""
+        self.cursor.execute('''CREATE TABLE quota (
+            id SERIAL PRIMARY KEY,
+            name TEXT NOT NULL,
+            concurrent_executions INT NOT NULL,
+            memory BIGINT NOT NULL,
+            cores INT NOT NULL
+        )''')
+        self.cursor.execute('''INSERT INTO quota (id, name, concurrent_executions, memory, cores) VALUES (DEFAULT, 'default', 5, 34359738368, 20)''')
+
+    def select(self, only_one=False, **kwargs):
+        """
+        Return a list of quotas.
+
+        :param only_one: only one result is expected
+        :type only_one: bool
+        :param kwargs: filter services based on their fields/columns
+        :return: one or more ports
+        """
+        q_base = 'SELECT * FROM quota'
+        if len(kwargs) > 0:
+            q = q_base + " WHERE "
+            filter_list = []
+            args_list = []
+            for key, value in kwargs.items():
+                filter_list.append('{} = %s'.format(key))
+                args_list.append(value)
+            q += ' AND '.join(filter_list)
+            query = self.cursor.mogrify(q, args_list)
+        else:
+            query = self.cursor.mogrify(q_base)
+
+        self.cursor.execute(query)
+        if only_one:
+            row = self.cursor.fetchone()
+            if row is None:
+                return None
+            return Quota(row, self)
+        else:
+            return [Quota(x, self) for x in self.cursor]
+
+    def update(self, record_id, **kwargs):
+        """Update the state of an existing quota."""
+        arg_list = []
+        value_list = []
+        for key, value in kwargs.items():
+            arg_list.append('{} = %s'.format(key))
+            value_list.append(value)
+        set_q = ", ".join(arg_list)
+        value_list.append(record_id)
+        q_base = 'UPDATE quota SET ' + set_q + ' WHERE id=%s'
+        query = self.cursor.mogrify(q_base, value_list)
+        self.cursor.execute(query)
+        self.sql_manager.commit()
+
+    def insert(self, name, concurrent_executions, memory, cores):
+        """Adds a new quota to the state."""
+        query = self.cursor.mogrify('INSERT INTO quota (id, name, concurrent_executions, memory, cores) VALUES (DEFAULT, %s, %s, %s, %s) RETURNING id', (name, concurrent_executions, memory, cores))
+        self.cursor.execute(query)
+        self.sql_manager.commit()
+        return self.cursor.fetchone()[0]
+
+    def delete(self, record_id):
+        """Delete a quota from the state."""
+        query = 'UPDATE "user" SET quota_id = (SELECT id from quota WHERE name=\'default\') WHERE quota_id=%s'
+        self.cursor.execute(query, (record_id,))
+        query = "DELETE FROM quota WHERE id = %s"
+        self.cursor.execute(query, (record_id,))
+        self.sql_manager.commit()

zoe_lib/state/role.py

+# Copyright (c) 2017, Daniele Venzano
+#
+# Licensed 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.
+
+"""Interface to PostgresQL for Zoe state."""
+
+import logging
+
+from zoe_lib.state.base import BaseTable, BaseRecord
+
+log = logging.getLogger(__name__)
+
+
+class Role(BaseRecord):
+    """A role object describes the permissions of groups of users."""
+
+    def __init__(self, d, sql_manager):
+        super().__init__(d, sql_manager)
+
+        self.name = d['name']
+        self.can_see_status = d['can_see_status']
+        self.can_change_config = d['can_change_config']
+        self.can_operate_others = d['can_operate_others']
+        self.can_delete_executions = d['can_delete_executions']
+
+    def serialize(self):
+        """Generates a dictionary that can be serialized in JSON."""
+        return {
+            'id': self.id,
+            'name': self.name,
+            'can_see_status': self.can_see_status,
+            'can_change_config': self.can_change_config,
+            'can_operate_others': self.can_operate_others,
+            'can_delete_executions': self.can_delete_executions
+        }
+
+
+class RoleTable(BaseTable):
+    """Abstraction for the role table in the database."""
+    def __init__(self, sql_manager):
+        super().__init__(sql_manager, "role")
+
+    def create(self):
+        """Create the role table."""
+        self.cursor.execute('''CREATE TABLE role (
+            id SERIAL PRIMARY KEY,
+            name TEXT NOT NULL,
+            can_see_status BOOLEAN NOT NULL DEFAULT FALSE,
+            can_change_config BOOLEAN NOT NULL DEFAULT FALSE,
+            can_operate_others BOOLEAN NOT NULL DEFAULT FALSE,
+            can_delete_executions BOOLEAN NOT NULL DEFAULT FALSE
+        )''')
+        self.cursor.execute('''INSERT INTO role (id, name, can_see_status, can_change_config, can_operate_others, can_delete_executions) VALUES (DEFAULT, 'admin', TRUE, TRUE, TRUE, TRUE)''')
+        self.cursor.execute('''INSERT INTO role (id, name, can_see_status, can_change_config, can_operate_others, can_delete_executions) VALUES (DEFAULT, 'superuser', TRUE, FALSE, FALSE, FALSE)''')
+        self.cursor.execute('''INSERT INTO role (id, name, can_see_status, can_change_config, can_operate_others, can_delete_executions) VALUES (DEFAULT, 'user', FALSE , FALSE, FALSE, FALSE)''')
+
+    def select(self, only_one=False, **kwargs):
+        """
+        Return a list of roles.
+
+        :param only_one: only one result is expected
+        :type only_one: bool
+        :param kwargs: filter services based on their fields/columns
+        :return: one or more ports
+        """
+        q_base = 'SELECT * FROM role'
+        if len(kwargs) > 0:
+            q = q_base + " WHERE "
+            filter_list = []
+            args_list = []
+            for key, value in kwargs.items():
+                filter_list.append('{} = %s'.format(key))
+                args_list.append(value)
+            q += ' AND '.join(filter_list)
+            query = self.cursor.mogrify(q, args_list)
+        else:
+            query = self.cursor.mogrify(q_base)
+
+        self.cursor.execute(query)
+        if only_one:
+            row = self.cursor.fetchone()
+            if row is None:
+                return None
+            return Role(row, self)
+        else:
+            return [Role(x, self) for x in self.cursor]
+
+    def update(self, record_id, **kwargs):
+        """Update the state of an existing role."""
+        arg_list = []
+        value_list = []
+        for key, value in kwargs.items():
+            arg_list.append('{} = %s'.format(key))
+            value_list.append(value)
+        set_q = ", ".join(arg_list)
+        value_list.append(record_id)
+        q_base = 'UPDATE role SET ' + set_q + ' WHERE id=%s'
+        query = self.cursor.mogrify(q_base, value_list)
+        self.cursor.execute(query)
+        self.sql_manager.commit()
+
+    def insert(self, name):
+        """Adds a new role to the state."""
+        query = self.cursor.mogrify('INSERT INTO role (id, name) VALUES (DEFAULT, %s) RETURNING id', (name,))
+        self.cursor.execute(query)
+        self.sql_manager.commit()
+        return self.cursor.fetchone()[0]
+
+    def delete(self, role_id):
+        """Delete a role from the state."""
+        query = 'UPDATE "user" SET role_id = (SELECT id from role WHERE name=\'user\') WHERE role_id=%s'
+        self.cursor.execute(query, (role_id,))
+        query = "DELETE FROM role WHERE id = %s"
+        self.cursor.execute(query, (role_id,))
+        self.sql_manager.commit()

zoe_lib/state/sql_manager.py

@@ -27,6 +27,9 @@ import zoe_lib.exceptions
 from .service import ServiceTable
 from .execution import ExecutionTable
 from .port import PortTable
+from .quota import QuotaTable
+from .user import UserTable
+from .role import RoleTable
 
 log = logging.getLogger(__name__)
 
@@ -36,7 +39,7 @@ psycopg2.extensions.register_adapter(dict, psycopg2.extras.Json)
 class SQLManager:
     """The SQLManager class, should be used as a singleton."""
     def __init__(self, conf):
-        self.user = conf.dbuser
+        self.dbuser = conf.dbuser
         self.password = conf.dbpass
         self.host = conf.dbhost
         self.port = conf.dbport
@@ -47,7 +50,7 @@ class SQLManager:
 
     def _connect(self):
         dsn = 'dbname=' + self.dbname + \
-              ' user=' + self.user + \
+              ' user=' + self.dbuser + \
               ' password=' + self.password + \
               ' host=' + self.host + \
               ' port=' + str(self.port)
@@ -83,10 +86,28 @@ class SQLManager:
         """Access the port state."""
         return PortTable(self)
 
+    @property
+    def quota(self) -> QuotaTable:
+        """Access the quota state."""
+        return QuotaTable(self)
+
+    @property
+    def role(self) -> RoleTable:
+        """Access the role state."""
+        return RoleTable(self)
+
+    @property
+    def user(self) -> UserTable:
+        """Access the user state."""
+        return UserTable(self)
+
     def _create_tables(self):
         self.executions.create()
         self.services.create()
         self.ports.create()
+        self.quota.create()
+        self.role.create()
+        self.user.create()
 
     def init_db(self, force=False):
         """DB init entrypoint."""

zoe_lib/state/user.py

+# Copyright (c) 2017, Daniele Venzano
+#
+# Licensed 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.
+
+"""Interface to PostgresQL for Zoe state."""
+
+import logging
+
+from zoe_lib.state.base import BaseRecord, BaseTable
+
+log = logging.getLogger(__name__)
+
+
+class User(BaseRecord):
+    """An user object describes a Zoe user."""
+
+    def __init__(self, d, sql_manager):
+        super().__init__(d, sql_manager)
+
+        self.username = d['username']
+        self.fs_uid = d['fs_uid']
+        self.role = d['role']
+        self.email = d['email']
+        self.priority = d['priority']
+        self.enabled = d['enabled']
+        self.quota_id = d['quota_id']
+
+    def serialize(self):
+        """Generates a dictionary that can be serialized in JSON."""
+        return {
+            'id': self.id,
+            'username': self.username,
+            'fs_uid': self.fs_uid,
+            'role': self.role,
+            'email': self.email,
+            'priority': self.priority,
+            'enabled': self.enabled,
+            'quota_id': self.quota_id
+        }
+
+    def set_email(self, new_email: str):
+        """Update the email address for this user."""
+        self.email = new_email
+        self.sql_manager.user_update(self.id, email=new_email)
+
+    def set_priority(self, new_priority: int):
+        """Update the priority for this user."""
+        self.priority = new_priority
+        self.sql_manager.user_update(self.id, priority=new_priority)
+
+    def set_enabled(self, enable: bool):
+        """Enable or disable a user."""
+        self.enabled = enable
+        self.sql_manager.user_update(self.id, enabled=enable)
+
+    def get_quota(self):
+        """Get the quota for this user."""
+        return self.sql_manager.quota_list(only_one=True, id=self.quota_id)
+
+    def set_quota(self, quota_id):
+        """Set a different quota for this user."""
+        self.quota_id = quota_id
+        self.sql_manager.user_update(self.id, quota_id=quota_id)
+
+    def set_role(self, role):
+        """Set a new role for this user."""
+        self.role = role
+        self.sql_manager.user_update(self.id, role=role)
+
+
+class UserTable(BaseTable):
+    """Abstraction for the user table in the database."""
+    def __init__(self, sql_manager):
+        super().__init__(sql_manager, "user")
+
+    def create(self):
+        """Create the user table."""
+        self.cursor.execute('''CREATE TABLE "user" (
+            id SERIAL PRIMARY KEY,
+            username TEXT NOT NULL,
+            fs_uid INT NOT NULL,
+            email TEXT,
+            priority SMALLINT NOT NULL DEFAULT 0,
+            enabled BOOLEAN NOT NULL DEFAULT TRUE,
+            auth_source TEXT NOT NULL,
+            role_id INT REFERENCES role,
+            quota_id INT REFERENCES quota
+        )''')
+        self.cursor.execute('CREATE UNIQUE INDEX users_username_uindex ON "user" (username)')
+
+    def select(self, only_one=False, **kwargs):
+        """
+        Return a list of users.
+
+        :param only_one: only one result is expected
+        :type only_one: bool
+        :param kwargs: filter services based on their fields/columns
+        :return: one or more ports
+        """
+        q_base = 'SELECT * FROM "user"'
+        if len(kwargs) > 0:
+            q = q_base + " WHERE "
+            filter_list = []
+            args_list = []
+            for key, value in kwargs.items():
+                filter_list.append('{} = %s'.format(key))
+                args_list.append(value)
+            q += ' AND '.join(filter_list)
+            query = self.cursor.mogrify(q, args_list)
+        else:
+            query = self.cursor.mogrify(q_base)
+
+        self.cursor.execute(query)
+        if only_one:
+            row = self.cursor.fetchone()
+            if row is None:
+                return None
+            return User(row, self)
+        else:
+            return [User(x, self) for x in self.cursor]
+
+    def update(self, port_id, **kwargs):
+        """Update the state of an user port."""
+        arg_list = []
+        value_list = []
+        for key, value in kwargs.items():
+            arg_list.append('{} = %s'.format(key))
+            value_list.append(value)
+        set_q = ", ".join(arg_list)
+        value_list.append(port_id)
+        q_base = 'UPDATE "user" SET ' + set_q + ' WHERE id=%s'
+        query = self.cursor.mogrify(q_base, value_list)
+        self.cursor.execute(query)
+        self.sql_manager.commit()
+
+    def insert(self, username, fs_uid, role, auth_source):
+        """Adds a new user to the state."""
+        query = self.cursor.mogrify('INSERT INTO "user" (id, username, fs_uid, email, priority, enabled, auth_source, role_id, quota_id) VALUES (DEFAULT, %s, %s, NULL, DEFAULT, DEFAULT, %s, (SELECT id FROM role WHERE name=%s), (SELECT id FROM quota WHERE name=\'default\')) RETURNING id', (username, fs_uid, auth_source, role))
+        self.cursor.execute(query)
+        self.sql_manager.commit()
+        return self.cursor.fetchone()[0]

zoe_lib/version.py

@@ -18,4 +18,4 @@
 ZOE_VERSION = '2018.03-beta'
 ZOE_API_VERSION = '0.7'
 ZOE_APPLICATION_FORMAT_VERSION = 3
-SQL_SCHEMA_VERSION = 6  # ---> Increment this value every time the SQL schema changes !!! <---
+SQL_SCHEMA_VERSION = 7  # ---> Increment this value every time the SQL schema changes !!! <---

zoe_master/backends/common.py

@@ -39,12 +39,13 @@ def gen_environment(execution: Execution, service: Service, env_subst_dict: Dict
 
     env_list.append(('EXECUTION_ID', str(execution.id)))
     env_list.append(('DEPLOY_NAME', get_conf().deployment_name))
-    env_list.append(('UID', execution.user_id))
+    env_list.append(('UID', execution.owner.fs_uid))
+    env_list.append(('USER', execution.owner.username))
     env_list.append(('SERVICE_NAME', service.name))
-    env_list.append(('PROXY_PATH', get_conf().proxy_path))
 
-    wk_vol = ZoeFSWorkspace().get(execution.user_id)
+    wk_vol = ZoeFSWorkspace().get(execution.owner.username)
     env_list.append(('ZOE_WORKSPACE', wk_vol.mount_point))
+    env_list.append(('HOME', wk_vol.mount_point))