Merge main

app/app.py

@@ -1,11 +1,15 @@
 import os
-from flask import Flask, render_template, jsonify, request, abort, send_from_directory
+from functools import wraps
+
+from flask import Flask, abort, jsonify, render_template, request
 from flask_swagger import swagger
 from flask_swagger_ui import get_swaggerui_blueprint
-from pkg_resources import resource_filename
-from .api_keys import Database
-from app.language import detect_languages, transliterate
+
 from app import flood
+from app.language import detect_languages, transliterate
+
+from .api_keys import Database
+
 
 def get_json_dict(request):
     d = request.get_json()
@@ -13,11 +17,12 @@ def get_json_dict(request):
         abort(400, description="Invalid JSON format")
     return d
 
+
 def get_remote_address():
     if request.headers.getlist("X-Forwarded-For"):
         ip = request.headers.getlist("X-Forwarded-For")[0]
     else:
-        ip = request.remote_addr or '127.0.0.1'
+        ip = request.remote_addr or "127.0.0.1"
 
     return ip
 
@@ -58,47 +63,91 @@ def get_routes_limits(default_req_limit, daily_req_limit, api_keys_db):
 
     return res
 
+
 def create_app(args):
     from app.init import boot
+
     boot(args.load_only)
 
     from app.language import languages
+
     app = Flask(__name__)
 
     if args.debug:
-        app.config['TEMPLATES_AUTO_RELOAD'] = True
+        app.config["TEMPLATES_AUTO_RELOAD"] = True
 
     # Map userdefined frontend languages to argos language object.
     if args.frontend_language_source == "auto":
-        frontend_argos_language_source = type('obj', (object,), {
-            'code': 'auto',
-            'name': 'Auto Detect'
-        })
+        frontend_argos_language_source = type(
+            "obj", (object,), {"code": "auto", "name": "Auto Detect"}
+        )
     else:
-        frontend_argos_language_source = next(iter([l for l in languages if l.code == args.frontend_language_source]), None)
+        frontend_argos_language_source = next(
+            iter([l for l in languages if l.code == args.frontend_language_source]),
+            None,
+        )
 
-    frontend_argos_language_target = next(iter([l for l in languages if l.code == args.frontend_language_target]), None)
+    frontend_argos_language_target = next(
+        iter([l for l in languages if l.code == args.frontend_language_target]), None
+    )
 
     # Raise AttributeError to prevent app startup if user input is not valid.
     if frontend_argos_language_source is None:
-        raise AttributeError(f"{args.frontend_language_source} as frontend source language is not supported.")
+        raise AttributeError(
+            f"{args.frontend_language_source} as frontend source language is not supported."
+        )
     if frontend_argos_language_target is None:
-        raise AttributeError(f"{args.frontend_language_target} as frontend target language is not supported.")
+        raise AttributeError(
+            f"{args.frontend_language_target} as frontend target language is not supported."
+        )
+
+    api_keys_db = None
 
     if args.req_limit > 0 or args.api_keys or args.daily_req_limit > 0:
+        api_keys_db = Database() if args.api_keys else None
+
         from flask_limiter import Limiter
+
         limiter = Limiter(
             app,
             key_func=get_remote_address,
-            default_limits=get_routes_limits(args.req_limit, args.daily_req_limit, Database() if args.api_keys else None)
+            default_limits=get_routes_limits(
+                args.req_limit, args.daily_req_limit, api_keys_db
+            ),
         )
     else:
-      from .no_limiter import Limiter
-      limiter = Limiter()
-    
+        from .no_limiter import Limiter
+
+        limiter = Limiter()
+
     if args.req_flood_threshold > 0:
         flood.setup(args.req_flood_threshold)
 
+    def access_check(f):
+        @wraps(f)
+        def func(*a, **kw):
+            if flood.is_banned(get_remote_address()):
+                abort(403, description="Too many request limits violations")
+
+            if args.api_keys and args.require_api_key_origin:
+                if request.is_json:
+                    json = get_json_dict(request)
+                    ak = json.get("api_key")
+                else:
+                    ak = request.values.get("api_key")
+
+                if (
+                    api_keys_db.lookup(ak) is None and request.headers.get("Origin") != args.require_api_key_origin
+                ):
+                    abort(
+                        403,
+                        description="Please contact the server operator to obtain an API key",
+                    )
+
+            return f(*a, **kw)
+
+        return func
+
     @app.errorhandler(400)
     def invalid_api(e):
         return jsonify({"error": str(e.description)}), 400
@@ -116,18 +165,23 @@ def create_app(args):
     def denied(e):
         return jsonify({"error": str(e.description)}), 403
 
-
     @app.route("/")
     @limiter.exempt
     def index():
-        return render_template('index.html', gaId=args.ga_id, frontendTimeout=args.frontend_timeout, api_keys=args.api_keys, web_version=os.environ.get('LT_WEB') is not None)
+        return render_template(
+            "index.html",
+            gaId=args.ga_id,
+            frontendTimeout=args.frontend_timeout,
+            api_keys=args.api_keys,
+            web_version=os.environ.get("LT_WEB") is not None,
+        )
 
-    @app.route("/javascript-licenses", methods=['GET'])
+    @app.route("/javascript-licenses", methods=["GET"])
     @limiter.exempt
     def javascript_licenses():
-        return render_template('javascript-licenses.html')
+        return render_template("javascript-licenses.html")
 
-    @app.route("/languages", methods=['GET', 'POST'])
+    @app.route("/languages", methods=["GET", "POST"])
     @limiter.exempt
     def langs():
         """
@@ -160,21 +214,23 @@ def create_app(args):
                   type: string
                   description: Reason for slow down
         """
-        return jsonify([{'code': l.code, 'name': l.name} for l in languages])
+        return jsonify([{"code": l.code, "name": l.name} for l in languages])
 
     # Add cors
     @app.after_request
     def after_request(response):
-        response.headers.add('Access-Control-Allow-Origin','*')
-        response.headers.add('Access-Control-Allow-Headers', "Authorization, Content-Type")
-        response.headers.add('Access-Control-Expose-Headers', "Authorization")
-        response.headers.add('Access-Control-Allow-Methods', "GET, POST")
-        response.headers.add('Access-Control-Allow-Credentials', "true")
-        response.headers.add('Access-Control-Max-Age', 60 * 60 * 24 * 20)
+        response.headers.add("Access-Control-Allow-Origin", "*")
+        response.headers.add(
+            "Access-Control-Allow-Headers", "Authorization, Content-Type"
+        )
+        response.headers.add("Access-Control-Expose-Headers", "Authorization")
+        response.headers.add("Access-Control-Allow-Methods", "GET, POST")
+        response.headers.add("Access-Control-Allow-Credentials", "true")
+        response.headers.add("Access-Control-Max-Age", 60 * 60 * 24 * 20)
         return response
 
-
-    @app.route("/translate", methods=['POST'])
+    @app.route("/translate", methods=["POST"])
+    @access_check
     def translate():
         """
         Translate text from a language to another
@@ -262,14 +318,11 @@ def create_app(args):
                   type: string
                   description: Error message
         """
-        if flood.is_banned(get_remote_address()):
-            abort(403, description="Too many request limits violations")
-
         if request.is_json:
             json = get_json_dict(request)
-            q = json.get('q')
-            source_lang = json.get('source')
-            target_lang = json.get('target')
+            q = json.get("q")
+            source_lang = json.get("source")
+            target_lang = json.get("target")
         else:
             q = request.values.get("q")
             source_lang = request.values.get("source")
@@ -285,20 +338,28 @@ def create_app(args):
         batch = isinstance(q, list)
 
         if batch and args.batch_limit != -1:
-          batch_size = len(q)
-          if args.batch_limit < batch_size:
-            abort(400, description="Invalid request: Request (%d) exceeds text limit (%d)" % (batch_size, args.batch_limit))
+            batch_size = len(q)
+            if args.batch_limit < batch_size:
+                abort(
+                    400,
+                    description="Invalid request: Request (%d) exceeds text limit (%d)"
+                    % (batch_size, args.batch_limit),
+                )
 
         if args.char_limit != -1:
             if batch:
-              chars = sum([len(text) for text in q])
+                chars = sum([len(text) for text in q])
             else:
-              chars = len(q)
+                chars = len(q)
 
             if args.char_limit < chars:
-              abort(400, description="Invalid request: Request (%d) exceeds character limit (%d)" % (chars, args.char_limit))
+                abort(
+                    400,
+                    description="Invalid request: Request (%d) exceeds character limit (%d)"
+                    % (chars, args.char_limit),
+                )
 
-        if source_lang == 'auto':
+        if source_lang == "auto":
             candidate_langs = detect_languages(q)
 
             if args.debug:
@@ -320,14 +381,30 @@ def create_app(args):
         translator = src_lang.get_translation(tgt_lang)
 
         try:
-          if batch:
-            return jsonify({"translatedText": [translator.translate(transliterate(text, target_lang=source_lang)) for text in q] })
-          else:
-            return jsonify({"translatedText": translator.translate(transliterate(q, target_lang=source_lang)) })
+            if batch:
+                return jsonify(
+                    {
+                        "translatedText": [
+                            translator.translate(
+                                transliterate(text, target_lang=source_lang)
+                            )
+                            for text in q
+                        ]
+                    }
+                )
+            else:
+                return jsonify(
+                    {
+                        "translatedText": translator.translate(
+                            transliterate(q, target_lang=source_lang)
+                        )
+                    }
+                )
         except Exception as e:
             abort(500, description="Cannot translate text: %s" % str(e))
 
-    @app.route("/detect", methods=['POST'])
+    @app.route("/detect", methods=["POST"])
+    @access_check
     def detect():
         """
         Detect the language of a single text
@@ -377,7 +454,7 @@ def create_app(args):
               properties:
                 error:
                   type: string
-                  description: Error message          
+                  description: Error message
           500:
             description: Detection error
             schema:
@@ -411,7 +488,7 @@ def create_app(args):
 
         if request.is_json:
             json = get_json_dict(request)
-            q = json.get('q')
+            q = json.get("q")
         else:
             q = request.values.get("q")
 
@@ -420,7 +497,6 @@ def create_app(args):
 
         return jsonify(detect_languages(q))
 
-
     @app.route("/frontend/settings")
     @limiter.exempt
     def frontend_settings():
@@ -464,30 +540,37 @@ def create_app(args):
                           type: string
                           description: Human-readable language name (in English)
         """
-        return jsonify({'charLimit': args.char_limit,
-                        'frontendTimeout': args.frontend_timeout,
-                        'language': {
-                            'source': {'code': frontend_argos_language_source.code, 'name': frontend_argos_language_source.name},
-                            'target': {'code': frontend_argos_language_target.code, 'name': frontend_argos_language_target.name}}
-                       })
+        return jsonify(
+            {
+                "charLimit": args.char_limit,
+                "frontendTimeout": args.frontend_timeout,
+                "language": {
+                    "source": {
+                        "code": frontend_argos_language_source.code,
+                        "name": frontend_argos_language_source.name,
+                    },
+                    "target": {
+                        "code": frontend_argos_language_target.code,
+                        "name": frontend_argos_language_target.name,
+                    },
+                },
+            }
+        )
 
     swag = swagger(app)
-    swag['info']['version'] = "1.2"
-    swag['info']['title'] = "LibreTranslate"
+    swag["info"]["version"] = "1.2"
+    swag["info"]["title"] = "LibreTranslate"
 
     @app.route("/spec")
     @limiter.exempt
     def spec():
         return jsonify(swag)
 
-    SWAGGER_URL = '/docs'  # URL for exposing Swagger UI (without trailing '/')
-    API_URL = '/spec'
+    SWAGGER_URL = "/docs"  # URL for exposing Swagger UI (without trailing '/')
+    API_URL = "/spec"
 
     # Call factory function to create our blueprint
-    swaggerui_blueprint = get_swaggerui_blueprint(
-        SWAGGER_URL,
-        API_URL
-    )
+    swaggerui_blueprint = get_swaggerui_blueprint(SWAGGER_URL, API_URL)
 
     app.register_blueprint(swaggerui_blueprint)