Merge pull request #797 from cquintana92/feature/only-allow-relative-redirects

Only allow relative and controlled redirects

app/auth/views/activate.py

@@ -7,6 +7,7 @@ from app.db import Session
 from app.extensions import limiter
 from app.log import LOG
 from app.models import ActivationCode
+from app.utils import sanitize_next_url
 
 
 @auth_bp.route("/activate", methods=["GET", "POST"])
@@ -58,7 +59,7 @@ def activate():
 
     # The activation link contains the original page, for ex authorize page
     if "next" in request.args:
-        next_url = request.args.get("next")
+        next_url = sanitize_next_url(request.args.get("next"))
         LOG.d("redirect user to %s", next_url)
         return redirect(next_url)
     else:

app/auth/views/facebook.py

@@ -13,7 +13,7 @@ from app.db import Session
 from app.log import LOG
 from app.models import User, SocialAuth
 from .login_utils import after_login
-from ...utils import sanitize_email
+from ...utils import sanitize_email, sanitize_next_url
 
 _authorization_base_url = "https://www.facebook.com/dialog/oauth"
 _token_url = "https://graph.facebook.com/oauth/access_token"
@@ -30,7 +30,7 @@ def facebook_login():
     # to avoid flask-login displaying the login error message
     session.pop("_flashes", None)
 
-    next_url = request.args.get("next")
+    next_url = sanitize_next_url(request.args.get("next"))
 
     # Facebook does not allow to append param to redirect_uri
     # we need to pass the next url by session

app/auth/views/login.py

@@ -8,7 +8,7 @@ from app.auth.views.login_utils import after_login
 from app.extensions import limiter
 from app.log import LOG
 from app.models import User
-from app.utils import sanitize_email
+from app.utils import sanitize_email, sanitize_next_url
 
 
 class LoginForm(FlaskForm):
@@ -21,7 +21,7 @@ class LoginForm(FlaskForm):
     "10/minute", deduct_when=lambda r: hasattr(g, "deduct_limit") and g.deduct_limit
 )
 def login():
-    next_url = request.args.get("next")
+    next_url = sanitize_next_url(request.args.get("next"))
 
     if current_user.is_authenticated:
         if next_url:

app/config.py

@@ -4,7 +4,7 @@ import socket
 import string
 import subprocess
 from ast import literal_eval
-from typing import Callable
+from typing import Callable, List
 from urllib.parse import urlparse
 
 from dotenv import load_dotenv
@@ -418,3 +418,14 @@ PHONE_PROVIDER_2_SECRET = os.environ.get("PHONE_PROVIDER_2_SECRET")
 ZENDESK_HOST = os.environ.get("ZENDESK_HOST")
 ZENDESK_API_TOKEN = os.environ.get("ZENDESK_API_TOKEN")
 ZENDESK_ENABLED = "ZENDESK_ENABLED" in os.environ
+
+
+def get_allowed_redirect_domains() -> List[str]:
+    allowed_domains = sl_getenv("ALLOWED_REDIRECT_DOMAINS", list)
+    if allowed_domains:
+        return allowed_domains
+    parsed_url = urlparse(URL)
+    return [parsed_url.hostname]
+
+
+ALLOWED_REDIRECT_DOMAINS = get_allowed_redirect_domains()

app/utils.py

@@ -3,10 +3,11 @@ import string
 import time
 import urllib.parse
 from functools import wraps
+from typing import List, Optional
 
 from unidecode import unidecode
 
-from .config import WORDS_FILE_PATH
+from .config import WORDS_FILE_PATH, ALLOWED_REDIRECT_DOMAINS
 from .log import LOG
 
 with open(WORDS_FILE_PATH) as f:
@@ -74,6 +75,38 @@ def sanitize_email(email_address: str, not_lower=False) -> str:
     return email_address
 
 
+class NextUrlSanitizer:
+    @staticmethod
+    def sanitize(url: Optional[str], allowed_domains: List[str]) -> Optional[str]:
+        if not url:
+            return None
+        # Relative redirect
+        if url[0] == "/":
+            return url
+        return NextUrlSanitizer.__handle_absolute_redirect(url, allowed_domains)
+
+    @staticmethod
+    def __handle_absolute_redirect(
+        url: str, allowed_domains: List[str]
+    ) -> Optional[str]:
+        if not NextUrlSanitizer.__is_absolute_url(url):
+            # Unknown url, something like &next=something.example.com
+            return None
+        parsed = urllib.parse.urlparse(url)
+        if parsed.hostname in allowed_domains:
+            return url
+        # Not allowed domain
+        return None
+
+    @staticmethod
+    def __is_absolute_url(url: str) -> bool:
+        return url.startswith(("http://", "https://"))
+
+
+def sanitize_next_url(url: Optional[str]) -> Optional[str]:
+    return NextUrlSanitizer.sanitize(url, ALLOWED_REDIRECT_DOMAINS)
+
+
 def query2str(query):
     """Useful utility method to print out a SQLAlchemy query"""
     return query.statement.compile(compile_kwargs={"literal_binds": True})

example.env

@@ -171,4 +171,7 @@ DISABLE_ONBOARDING=true
 
 # TEMP_DIR = /tmp
 
-#ALIAS_AUTOMATIC_DISABLE=true
+#ALIAS_AUTOMATIC_DISABLE=true
+
+# domains that can be present in the &next= section when using absolute urls
+ALLOWED_REDIRECT_DOMAINS=[]

tests/test.env

@@ -51,4 +51,5 @@ FACEBOOK_CLIENT_SECRET=to_fill
 
 PGP_SENDER_PRIVATE_KEY_PATH=local_data/private-pgp.asc
 
-ALIAS_AUTOMATIC_DISABLE=true
+ALIAS_AUTOMATIC_DISABLE=true
+ALLOWED_REDIRECT_DOMAINS=["test.simplelogin.local"]

tests/test_utils.py

@@ -1,4 +1,5 @@
-from app.utils import random_string, random_words
+from app.config import ALLOWED_REDIRECT_DOMAINS
+from app.utils import random_string, random_words, sanitize_next_url
 
 
 def test_random_words():
@@ -9,3 +10,24 @@ def test_random_words():
 def test_random_string():
     s = random_string()
     assert len(s) > 0
+
+
+def test_sanitize_url():
+    cases = [
+        {"url": None, "expected": None},
+        {"url": "", "expected": None},
+        {"url": "http://unknown.domain", "expected": None},
+        {"url": "https://badzone.org", "expected": None},
+        {"url": "/", "expected": "/"},
+        {"url": "/auth", "expected": "/auth"},
+        {"url": "/some/path", "expected": "/some/path"},
+    ]
+
+    for domain in ALLOWED_REDIRECT_DOMAINS:
+        cases.append({"url": f"http://{domain}", "expected": f"http://{domain}"})
+        cases.append({"url": f"https://{domain}", "expected": f"https://{domain}"})
+        cases.append({"url": domain, "expected": None})
+
+    for case in cases:
+        res = sanitize_next_url(case["url"])
+        assert res == case["expected"]