path: root/server/python_session.py

# CodeQ: an online programming tutor.
# Copyright (C) 2015 UL FRI
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU Affero General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option) any
# later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
# details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

from fcntl import fcntl, F_GETFL, F_SETFL
import io
import multiprocessing
import os
import pickle
import selectors
import signal
import subprocess
import sys
import threading
import time

import server
import server.user_session
from db.models import Problem

__all__ = ['PythonSession']

class PythonSession(server.LanguageSession):
    """Abstracts a Python session.
    Only public methods are available to the outside world due to the use of multiprocessing managers.
    Therefore prefix any private methods with an underscore (_).
    No properties are accessible; use getters and setters instead.
    Values are passed by value instead of by reference (deep copy!).
    """
    def __init__(self, output_cb=None):
        self._access_lock = threading.Lock()
        self._sent_hints = []

        self._notifier, receiver = multiprocessing.Pipe()
        self._interpreter = threading.Thread(target=_interpreter,
                                         kwargs={'control': receiver, 'callback': output_cb})
        self._interpreter.start()

    def run(self, code=None, inputs=None, timeout=1.0):
        # Launch processes.
        futures = []
        for expr, stdin in inputs:
            p = subprocess.Popen(proc_args + ['run'],
                    stdin=subprocess.PIPE,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.DEVNULL)
            data = pickle.dumps((code, expr, stdin))
            p.stdin.write(data)
            futures.append(p)

        # Wait for results.
        results = []
        start = time.monotonic()
        for proc in futures:
            now = time.monotonic()
            real_timeout = max(0, timeout - (now - start))
            try:
                stdout, _ = proc.communicate(timeout=real_timeout)
                result = pickle.loads(stdout)
                results.append(result)
            except subprocess.TimeoutExpired:
                results.append((None, '', '', 'timed out'))
            except EOFError:
                results.append((None, '', '', 'sandbox violation'))
            if proc.poll() is None:
                try:
                    proc_kill(proc, signal.SIGKILL)
                except:
                    pass

        return results

    def exec(self, program):
        self._notifier.send(('exec', program))

    def push(self, stdin):
        self._notifier.send(('push', stdin))

    def stop(self):
        self._notifier.send(('stop', None))

    def destroy(self):
        self._notifier.send(('done', None))

    def __del__(self):
        self.destroy()

    # TODO remove this
    def hint(self, sid, problem_id, program):
        return []

    def test(self, sid, problem_id, program):
        session = server.user_session.get_session_by_id(sid)
        p = Problem.get(id=problem_id)

        language_module = server.problems.load_language(p.language, 'common')
        problem_module = server.problems.load_problem(p.language, p.group, p.identifier, 'common')

        # experiment support for allowing none/automatic/manual/all hints
        # descriptor: {'id': 'hints', 'group': 'none|automatic|manual|all'}
        allowed_hints = 'all'
        for experiment in session.get_experiments():
            if experiment.get('id') == 'hints':
                allowed_hints = experiment.get('group')
                break

        # check if the program is correct
        passed, msgs = problem_module.test(self.run, program)
        if passed:
            session.update_solution(problem_id, solution=program, done=True)
        else:
            hints = []
            # manually defined problem-specific hints
            if not hints and hasattr(language_module, 'hint'):
                hints = language_module.hint(self.run, program)

            # generic language hints (style etc.)
            if not hints and hasattr(problem_module, 'hint') and \
                    allowed_hints in ('all', 'manual'):
                hints = problem_module.hint(self.run, program)

            if hints:
                msgs.extend(hints)

        self._instantiate_and_save_hints(language_module, problem_module, msgs)
        return msgs

    # Add hint parameters (such as message index) based on hint class. Append
    # the finalized hints to the list of sent hints.
    def _instantiate_and_save_hints(self, language_mod, problem_mod, hints):
        with self._access_lock:
            for hint in hints:
                for mod in [language_mod, problem_mod]:
                    if hasattr(mod, 'hint_type') and hint['id'] in mod.hint_type:
                        hint_type = mod.hint_type[hint['id']]
                        hint_type.instantiate(hint, self._sent_hints)
            self._sent_hints.extend(hints)

# If the sandbox wrapper exists, use it to switch to user "nobody" and enforce
# additional limits. Unless the daemon is running as root we are not able to
# signal nobody's PIDs, so switch user again for the killing.
_basedir = os.path.split(os.path.dirname(os.path.realpath(__file__)))[0]
_script = os.path.join(_basedir, 'python', 'runner', 'interpreter.py')
_sandbox = os.path.join(_basedir, 'python', 'runner', 'sandbox')
if os.path.exists(_sandbox):
    newuser = 'nobody'  # TODO make this configurable
    proc_args = [_sandbox, newuser, _script]
    proc_kill = lambda proc, sig: subprocess.call([_sandbox, newuser, 'kill', '-{}'.format(sig), str(proc.pid)])
else:
    proc_args = [_script]
    proc_kill = lambda proc, sig: proc.send_signal(sig)

def _interpreter(control, callback):
    done = False
    proc = None

    # Remember how much text and how many newlines we received this second; if
    # it is too much, kill the interpreter.
    # TODO this is a hack to prevent the JS console from becoming unresponsive,
    # it should be fixed there.
    now = 0
    length = newlines = 0

    selector = selectors.DefaultSelector()

    def command(conn):
        nonlocal proc, done
        # Get a control command.
        try:
            cmd, data = conn.recv()
            if cmd == 'exec':
                exec_str = 'exec("""\\\n{}\n""")\n'.format(data.replace('"', '\\"'))
                proc.stdin.write(exec_str.encode('utf-8'))
                proc.stdin.flush()
            elif cmd == 'push':
                proc.stdin.write(data.encode('utf-8'))
                proc.stdin.flush()
            elif cmd == 'stop':
                proc_kill(proc, signal.SIGINT)
            elif cmd == 'done':
                done = True
        except:
            pass

    def communicate(conn):
        nonlocal proc, callback, now, length, newlines
        if time.monotonic() - now > 1.0:
            length = newlines = 0
            now = time.monotonic()

        # Communicate with child process.
        data = proc.stdout.read1(1024)
        if data and length < 100000 and newlines < 1000:
            # NOTE this might fail if read() stops in the middle of utf8 sequence
            text = data.decode('utf-8')
            if text:
                callback(text)
                length += len(text)
                newlines += text.count('\n')
        else:
            selector.unregister(conn)
            if proc.poll() is None:
                # Process has not terminated yet, make sure it does.
                proc_kill(proc, signal.SIGKILL)
            proc = None
            length = newlines = 0
            callback('Interpreter restarted.\n')

    selector.register(control, selectors.EVENT_READ, command)
    while not done:
        # Ensure the interpreter process is running.
        if proc is None:
            proc = subprocess.Popen(proc_args,
                    stdin=subprocess.PIPE,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.STDOUT)
            # Set the non-blocking flag for stdout.
            flags = fcntl(proc.stdout.fileno(), F_GETFL)
            fcntl(proc.stdout.fileno(), F_SETFL, flags | os.O_NONBLOCK)
            selector.register(proc.stdout, selectors.EVENT_READ, communicate)

        events = selector.select()
        for key, mask in events:
            if mask & selectors.EVENT_READ:
                try:
                    key.data(key.fileobj)
                except:
                    pass

    # We are done, kill the child.
    selector.close()
    if proc is not None:
        proc_kill(proc, signal.SIGKILL)

server.language_session_handlers['python'] = lambda user_session, problem_id, language_identifier, group_identifier, problem_identifier: PythonSession(lambda text: user_session.send({'event': 'terminal_output', 'text': text}))