#!/usr/bin/env python3 # Flask server for the application # Author: Lilian BESSON # Email: lilian DOT besson AT crans D O T org # Version: 1 # Date: 19-02-2021 # Web: https://github.com/Naereen/Peut-on-coder-avec-OCaml-Python-et-C-par-SMS.git # # Source : # https://www.fullstackpython.com/blog/respond-sms-text-messages-python-flask.html # https://gist.github.com/mattmakai/8ab434ccb604d3ba5bde817a183e0bde # # Use to check if password file is present import os.path # Use time get string for today current hour import time # to launch a random test import random # To parse the password from the message import re from pprint import pprint from typing import Tuple, Union # Use base64 to not keep plaintext files of the number, username and password in your home import base64 from flask import Flask, Response, request # https://www.twilio.com/docs/sms/tutorials/how-to-receive-and-reply-python from twilio.twiml.messaging_response import MessagingResponse # Turn to True if DEBUG mode # DEBUG mode for a Flask app means that it automatically reloads when this file changes! DEBUG = False DEBUG = True # To avoid risking a HUGE Twilio bill, by default the server stops as soon as one language has received more than MAX_SMSNUMBER requests by SMS. MAX_SMSNUMBER = 100 def today(): return time.strftime("%H:%M:%S %Y-%m-%d") # ======================== password ======================== # DONE: read from .password.b64 file def read_b64_file(name: str) -> Union[None, str]: """ Open the local file , read and decode (base64) and return its content. """ try: with open(name) as f: print(f"DEBUG: reading file '{name}'") content = base64.b64decode(f.readline()[:-1]) while content[-1] == '\n': content = content[:-1] content = content.decode() print(f"DEBUG: content read is\n{content}") content = content.replace('\n', '') return content except OSError: print(f"Error: unable to read the file '{name}' ...") return None PASSWORD_FILE = ".password.b64" PASSWORD = "1234" if not os.path.exists(PASSWORD_FILE): print(f"\nPassword file '{PASSWORD_FILE}' does not exist, please enter a password!") local_password = input("Password= ") # TODO real password input? Flemme! so not important with open(PASSWORD_FILE, "wb") as file: bytes_local_password = local_password.encode() b64_local_password = base64.encodebytes(bytes_local_password) # str_b64_local_password = b64_local_password.decode() file.write(b64_local_password) PASSWORD = read_b64_file(PASSWORD_FILE) if PASSWORD is None: PASSWORD = "1234" print(f"Using password = {PASSWORD}...") def has_password(message: str) -> bool: """ Checks for presence of a password in message, in the form of `pw:SOMEPASSWORDNOSPACE`.""" print(f"DEBUG: message = '{message}'") print(f"DEBUG: and PASSWORD = '{PASSWORD}'") res = re.search("pw:([^ ]+) ", message) return res is not None def parse_password(message: str) -> str: """ Returns password in message, in the form of `pw:SOMEPASSWORDNOSPACE`.""" print(f"DEBUG: message = '{message}'") print(f"DEBUG: and PASSWORD = '{PASSWORD}'") res = re.search("pw:([^ ]+) ", message) if res: password = res.group(0) print(f"DEBUG: found a matching password = '{password}'") password = password.replace("pw:", "", 1) password = password.rstrip() print(f"DEBUG: cleaned to password = '{password}'") return password return "" def check_password(password: str) -> bool: """ Checks if password in message is correct.""" print(f"DEBUG: checking = '{password}'") print(f"DEBUG: and PASSWORD = '{PASSWORD}'") return password == PASSWORD # ======================== safe execution part ======================== from safeExecuteCode import safe_execute_code, URL, SUPPORTED_LANGUAGES class FailedExecution(Exception): pass # TODO: be able to really execute code def execute_code(inputcode: str, language="python") -> Tuple[str, str, int, dict]: print(f"DEBUG: You sent me this {language} code:\n{inputcode}") stdout, stderr = "", "" exitcode = 0 json_result = dict() stdout = f"You sent me this {language} code:\n{inputcode}" try: json_result = safe_execute_code(inputcode, language=language) print(f"DEBUG: I got back this JSON result:") pprint(json_result) # DEBUG if not json_result["success"]: raise FailedExecution first_result = json_result if "compile" in json_result \ and "exitcode" in json_result["compile"] \ and json_result["compile"]["exitcode"] > 0: compile_result = json_result["compile"] stdout = compile_result["stdout"] stderr = compile_result["stderr"] exitcode = compile_result["exitcode"] # Example of a reply of failed compilation: # {'compile': {'exitcode': 1, # 'meta': {'cg-mem': 8036, # 'cg-oom-killed': 0, # 'csw-forced': 12, # 'csw-voluntary': 4, # 'exitcode': 2, # 'exitsig': 0, # 'exitsig-message': None, # 'killed': False, # 'max-rss': 16532, # 'message': 'Exited with error status 2', # 'status': 'RUNTIME_ERROR', # 'time': 0.014, # 'wall-time': 0.027}, # 'stderr': 'File "/box/source.ml", line 1, characters 74-77:\n' # 'Error: This expression has type float but an ' # 'expression was expected of type\n' # ' int\n', # 'stdout': ''}, # 'success': True} else: if "tests" in json_result: first_result = json_result["tests"][0] stdout = first_result["stdout"] stderr = first_result["stderr"] exitcode = first_result["exitcode"] # Example of a correct reply: # { # "success": true, # "tests": [ # { # "exitcode": 0, # "name": "test000", # "stderr": "", # "stdout": "42\n" # } # ] # } except Exception as e: print("Error:\n", e) stderr = f"Camisole VM was probably not available, check the configuration.\n$ curl {URL}/\n$ curl {URL}/system\n$ curl {URL}/languages" stderr += f"\n\nError: {e}" exitcode = 1 # only for DEBUG, kill the server and print the calltrace if DEBUG: raise e # now we are done, give this back to Flask API return stdout, stderr, exitcode, json_result from collections import defaultdict cellnumbers = defaultdict(lambda: 0) smsnumber = 0 def format_reply(language: str, stdout: str, stderr: str, exitcode=0, full_data=None) -> str: """ Format the reply to a nice message that can be printed or sent back by SMS.""" global cellnumbers, smsnumber cellnumbers[language] += 1 cellnumber = cellnumbers[language] smsnumber += 1 assert smsnumber <= MAX_SMSNUMBER, f"Error: cell number for language='{language}' reached the maximum {MAX_SMSNUMBER}, so stopping the server.\nThis is NOT a bug, it's a feature, to avoid huge Twilio bills!" if stderr and stdout: reply = f"""Time: {today()}\nOut[{cellnumber}] {stdout}\nError[{cellnumber}] exitcode={exitcode} : {stderr}""" elif not stderr and stdout: reply = f"""Time: {today()}\nOut[{cellnumber}] {stdout}""" elif stderr and not stdout: reply = f"""Time: {today()}\nError[{cellnumber}] exitcode={exitcode} : {stderr}""" else: # TODO: print compiler error if something failed? # {'success': True, # 'tests': [{'exitcode': 1, # 'meta': {'cg-mem': 2492, # 'cg-oom-killed': 0, # 'csw-forced': 10, # 'csw-voluntary': 66, # 'exitcode': 0, # 'exitsig': 0, # 'exitsig-message': None, # 'killed': True, # 'max-rss': 7136, # 'message': 'Time limit exceeded (wall clock)', # 'status': 'TIMED_OUT', # 'time': 0.017, # 'wall-time': 60.116}, # 'name': 'test000', # 'stderr': '', # 'stdout': ''}] if full_data and "tests" in full_data and "meta" in full_data["tests"][0]: warning_response = "(empty default warning response)" # TODO: meta_data = full_data["tests"][0]["meta"] if "message" in meta_data: warning_response = meta_data["message"] if "(wall clock)" in warning_response and "wall-time" in meta_data: wall_time = meta_data["wall-time"] warning_response = warning_response.replace("(wall clock)", f"(wall clock) (time = {wall_time})") if "(time clock)" in warning_response and "time" in meta_data: time_clock = meta_data["time"] warning_response = warning_response.replace("(time clock)", f"(time clock) (time = {time_clock})") reply = f"""Time: {today()}\nFailed compilation or execution, got this reply for cell number {cellnumber}\n{warning_response}""" else: reply = f"""Time: {today()}\nNo output or error for cell number {cellnumber}""" return reply # ============== Test the API for two main languages ============== import first_tests def test_backend() -> None: """ Test the API for two main languages.""" for first_test in first_tests.FIRST_TESTS: print(f"\nDEBUG: trying to use this JSON request:\n{first_test}") inputcode = first_test["inputcode"] language = first_test["language"] print(f"\nDEBUG: trying to execute this code in language={language}:\n{inputcode}") stdout, stderr, exitcode, _ = execute_code(inputcode, language=language) print(f"DEBUG: got an answer with stdout, stderr, exitcode:\n{stdout}\n{stderr}\n{exitcode}") reply = format_reply(language, stdout, stderr, full_data=first_test) print(f"DEBUG: got a reply:\n{reply}") # TODO: when starting the server, it's expected that all tests should work! if not "What happens with a exitcode = 1 ?" in reply: assert exitcode == 0 if __name__ == '__main__': test_backend() # ================== now the Flask app ================== app = Flask("Peut on coder avec OCaml Python et C par SMS ?") @app.route("/") def check_app() -> Tuple[Response, int]: # returns a detailed documentation of the app! str_languages = ", ".join(SUPPORTED_LANGUAGES) return Response("""

« Peut on coder avec OCaml Python et C par SMS ? »

It works! The local server is ready! TODO: test the SMS part now, when Twilio will give me my number!

Documentation of this API

This API has the following end-points, please try them! Now, go read the documentation to conclude your setup (in French, so far). Enjoy!
If all the API end-points work, and if you have a Twilio account, then start "ngrok" for HTTP tunneling, add the ngrok proxy address to your Twilio dashboard, and try to send some code to your Twilio phone number (using your own phone).

Format of SMS messages

Be aware that each SMS costs around 0.07€!

About this project

Flask app started at """ + str(today) + """.
This script and this documentation are distributed in open access according to the conditions of the licence MIT. © Lilian Besson, 2021. """), 200 # return list of supported languages @app.route("/languages") def app_route_language() -> Tuple[Response, int]: str_languages = ", ".join(SUPPORTED_LANGUAGES) return Response(f"List of supported languages are: {str_languages}"), 200 @app.route("/langages") def app_route_langage() -> Tuple[Response, int]: str_languages = ", ".join(SUPPORTED_LANGUAGES) return Response(f"La liste des langues prises en charge est : {str_languages}"), 200 @app.route("/test") def app_route_test() -> Tuple[Response, int]: return Response(f"""Open one of these links to test the code execution API: """), 200 # TODO factor these /test/XXX endpints @app.route("/test/python") def app_route_testpython() -> Tuple[Response, int]: language = "python" random_test = random.choice(first_tests.TESTS_PYTHON)["inputcode"] stdout, stderr, exitcode, _ = execute_code(random_test, language=language) reply = format_reply(language, stdout, stderr, exitcode=exitcode) return Response(reply), 200 @app.route("/test/ocaml") def app_route_testocaml() -> Tuple[Response, int]: language = "ocaml" random_test = random.choice(first_tests.TESTS_OCAML)["inputcode"] stdout, stderr, exitcode, _ = execute_code(random_test, language=language) reply = format_reply(language, stdout, stderr, exitcode=exitcode) return Response(reply), 200 @app.route("/test/c") def app_route_testc() -> Tuple[Response, int]: language = "c" random_test = random.choice(first_tests.TESTS_C)["inputcode"] stdout, stderr, exitcode, _ = execute_code(random_test, language=language) reply = format_reply(language, stdout, stderr, exitcode=exitcode) return Response(reply), 200 # ============ now the Twilio part ============ # def inbound_sms() -> str: @app.route("/twilio", methods=["POST"]) def inbound_sms() -> Tuple[Response, int]: response = MessagingResponse() # we get the SMS message from the request. we could also get the # "To" and the "From" phone number as well inbound_message = request.form.get("Body") print(f"DEBUG: {today()} received a new message:\n{inbound_message}") # we can now use the incoming message text in our Python application # test messages # TODO: remove when DEBUG is done (https://github.com/Naereen/Peut-on-coder-avec-OCaml-Python-et-C-par-SMS/issues/1) if inbound_message == "test": # DONE it works! 2021-02-21 04:33:03 print(f"DEBUG: sending back\nIt works!") response.message("It works!") elif inbound_message == "Hello": # DONE it works! 2021-02-21 04:33:03 print(f"DEBUG: sending back\nHello back to you from Python!") response.message("Hello back to you from Python!") elif inbound_message == "Bonjour": # DONE it works! 2021-02-21 04:33:03 print(f"DEBUG: sending back\nBien le bonjour depuis Python !") response.message("Bien le bonjour depuis Python !") # return list of supported languages elif inbound_message == "Languages?": # DONE it works! 2021-02-21 04:33:03 str_languages = ", ".join(SUPPORTED_LANGUAGES) print(f"DEBUG: sending back\nList of supported languages are: {str_languages}") response.message(f"List of supported languages are: {str_languages}") # return list of supported languages elif inbound_message == "Langages ?": # DONE it works! 2021-02-21 04:33:03 str_languages = ", ".join(SUPPORTED_LANGUAGES) print(f"DEBUG: sending back\nLa liste des langues prises en charge est : {str_languages}") response.message(f"La liste des langues prises en charge est : {str_languages}") # now for languages else: # DONE add a password if not has_password(inbound_message): print("ERROR: No password! Add a password by starting your SMS with pw:PASSWORD, with no space!") # DEBUG return Response("No password! Add a password by starting your SMS with pw:PASSWORD, with no space!"), 500 password = parse_password(inbound_message) print(f"DEBUG: extracted password to be: '{password}'") inbound_message = inbound_message.replace(f"pw:{password} ", "", 1) if not check_password(password): print("ERROR: Wrong password! Hint: password might be 1234, if the developper was stupid!") # DEBUG return Response("Wrong password! Hint: password might be 1234, if the developper was stupid!"), 500 for language in SUPPORTED_LANGUAGES: if inbound_message.startswith(f"{language}:"): print(f"\nDEBUG: ==> detected that the message uses language = {language}") inbound_message = inbound_message.replace(f"{language}:\n", "", 1).lstrip() print(f"DEBUG: cleaned inbound_message:\n{inbound_message}") inbound_message = inbound_message.replace(f"{language}:", "", 1).lstrip() print(f"DEBUG: cleaned inbound_message:\n{inbound_message}") stdout, stderr, exitcode, full_data = execute_code(inbound_message, language=language) print(f"DEBUG: code stdout:\n{stdout}") print(f"DEBUG: code stderr:\n{stderr}") print(f"DEBUG: code exitcode = {exitcode}") reply = format_reply(language, stdout, stderr, exitcode=exitcode, full_data=full_data) print(f"DEBUG: giving back this reply:\n{reply}") response.message(reply) break # default response else: response.message("Hi! Not quite sure what you meant, but okay.\nLanguage not recognized maybe?\nSent 'Languages?' to get list of languages\nSee https://github.com/Naereen/Peut-on-coder-avec-OCaml-Python-et-C-par-SMS for more information!\n(C) Lilian Besson, 2021, MIT Licensed") print(f"DEBUG: done for the reply, sending back with code=200 (success), this text:\n{str(response)}") return Response(str(response)), 200 if __name__ == "__main__": app.run(debug=DEBUG)