diff --git a/app/config.py b/app/config.py index aedbb743..f0bc9b07 100644 --- a/app/config.py +++ b/app/config.py @@ -8,7 +8,6 @@ from urllib.parse import urlparse from dotenv import load_dotenv - ROOT_DIR = os.path.abspath(os.path.dirname(os.path.dirname(__file__))) @@ -523,3 +522,5 @@ if ENABLE_ALL_REVERSE_ALIAS_REPLACEMENT: MAX_NB_REVERSE_ALIAS_REPLACEMENT = int( os.environ["MAX_NB_REVERSE_ALIAS_REPLACEMENT"] ) + +DISABLE_RATE_LIMIT = "DISABLE_RATE_LIMIT" in os.environ diff --git a/app/extensions.py b/app/extensions.py index 32c47b36..58c6799d 100644 --- a/app/extensions.py +++ b/app/extensions.py @@ -2,6 +2,8 @@ from flask_limiter import Limiter from flask_limiter.util import get_remote_address from flask_login import current_user, LoginManager +from app import config + login_manager = LoginManager() login_manager.session_protection = "strong" @@ -20,6 +22,12 @@ def __key_func(): # Setup rate limit facility limiter = Limiter(key_func=__key_func) + +@limiter.request_filter +def disable_rate_limit(): + return config.DISABLE_RATE_LIMIT + + # @limiter.request_filter # def ip_whitelist(): # # Uncomment line to test rate limit in dev environment diff --git a/tests/api/test_import_export.py b/tests/api/test_import_export.py index 0a490a39..67b0da1c 100644 --- a/tests/api/test_import_export.py +++ b/tests/api/test_import_export.py @@ -114,7 +114,7 @@ def test_import_no_mailboxes_no_domains(flask_client): "ebay@my-domain.com,Used on eBay", 'facebook@my-domain.com,"Used on Facebook, Instagram."', ] - file = File.create(path="/test", commit=True) + file = File.create(path=f"/{random_token()}", commit=True) batch_import = BatchImport.create(user_id=user.id, file_id=file.id, commit=True) import_from_csv(batch_import, user, alias_data) @@ -130,19 +130,18 @@ def test_import_no_mailboxes(flask_client): # Check start state assert len(Alias.filter_by(user_id=user.id).all()) == 1 # Onboarding alias + domain = random_domain() # Create domain - CustomDomain.create( - user_id=user.id, domain="my-domain.com", ownership_verified=True - ) + CustomDomain.create(user_id=user.id, domain=domain, ownership_verified=True) Session.commit() alias_data = [ "alias,note", - "ebay@my-domain.com,Used on eBay", - 'facebook@my-domain.com,"Used on Facebook, Instagram."', + f"ebay@{domain},Used on eBay", + f'facebook@{domain},"Used on Facebook, Instagram."', ] - file = File.create(path="/test", commit=True) + file = File.create(path=f"/{random_token()}", commit=True) batch_import = BatchImport.create(user_id=user.id, file_id=file.id) import_from_csv(batch_import, user, alias_data) @@ -163,7 +162,7 @@ def test_import_no_domains(flask_client): 'facebook@my-domain.com,"Used on Facebook, Instagram.",destination1@my-destination-domain.com destination2@my-destination-domain.com', ] - file = File.create(path="/test", commit=True) + file = File.create(path=f"/{random_token()}", commit=True) batch_import = BatchImport.create(user_id=user.id, file_id=file.id) import_from_csv(batch_import, user, alias_data) @@ -179,31 +178,29 @@ def test_import(flask_client): # Check start state assert len(Alias.filter_by(user_id=user.id).all()) == 1 # Onboarding alias + domain1 = random_domain() + domain2 = random_domain() # Create domains - CustomDomain.create( - user_id=user.id, domain="my-domain.com", ownership_verified=True - ) - CustomDomain.create( - user_id=user.id, domain="my-destination-domain.com", ownership_verified=True - ) + CustomDomain.create(user_id=user.id, domain=domain1, ownership_verified=True) + CustomDomain.create(user_id=user.id, domain=domain2, ownership_verified=True) Session.commit() # Create mailboxes mailbox1 = Mailbox.create( - user_id=user.id, email="destination@my-destination-domain.com", verified=True + user_id=user.id, email=f"destination@{domain2}", verified=True ) mailbox2 = Mailbox.create( - user_id=user.id, email="destination2@my-destination-domain.com", verified=True + user_id=user.id, email=f"destination2@{domain2}", verified=True ) Session.commit() alias_data = [ "alias,note,mailboxes", - "ebay@my-domain.com,Used on eBay,destination@my-destination-domain.com", - 'facebook@my-domain.com,"Used on Facebook, Instagram.",destination@my-destination-domain.com destination2@my-destination-domain.com', + f"ebay@{domain1},Used on eBay,destination@{domain2}", + f'facebook@{domain1},"Used on Facebook, Instagram.",destination@{domain2} destination2@{domain2}', ] - file = File.create(path="/test", commit=True) + file = File.create(path=f"/{random_token()}", commit=True) batch_import = BatchImport.create(user_id=user.id, file_id=file.id) import_from_csv(batch_import, user, alias_data) @@ -214,7 +211,7 @@ def test_import(flask_client): # aliases[0] is the onboarding alias, skip it # eBay alias - assert aliases[1].email == "ebay@my-domain.com" + assert aliases[1].email == f"ebay@{domain1}" assert len(aliases[1].mailboxes) == 1 # First one should be primary assert aliases[1].mailbox_id == mailbox1.id @@ -222,7 +219,7 @@ def test_import(flask_client): assert aliases[1].mailboxes[0] == mailbox1 # Facebook alias - assert aliases[2].email == "facebook@my-domain.com" + assert aliases[2].email == f"facebook@{domain1}" assert len(aliases[2].mailboxes) == 2 # First one should be primary assert aliases[2].mailbox_id == mailbox1.id diff --git a/tests/api/test_new_custom_alias.py b/tests/api/test_new_custom_alias.py index 1f1975d0..b2842a34 100644 --- a/tests/api/test_new_custom_alias.py +++ b/tests/api/test_new_custom_alias.py @@ -1,12 +1,13 @@ from flask import g +from app import config from app.alias_suffix import signer from app.alias_utils import delete_alias from app.config import EMAIL_DOMAIN, MAX_NB_EMAIL_FREE_PLAN from app.db import Session from app.models import Alias, CustomDomain, Mailbox, AliasUsedOn from app.utils import random_word -from tests.utils import login, random_domain +from tests.utils import login, random_domain, random_token def test_v2(flask_client): @@ -92,12 +93,14 @@ def test_full_payload(flask_client): suffix = f".{word}@{EMAIL_DOMAIN}" signed_suffix = signer.sign(suffix).decode() - assert AliasUsedOn.count() == 0 + prefix = random_token() + + assert AliasUsedOn.filter(AliasUsedOn.user_id == user.id).count() == 0 r = flask_client.post( "/api/v3/alias/custom/new?hostname=example.com", json={ - "alias_prefix": "prefix", + "alias_prefix": prefix, "signed_suffix": signed_suffix, "note": "test note", "mailbox_ids": [user.default_mailbox_id, mb.id], @@ -106,7 +109,7 @@ def test_full_payload(flask_client): ) assert r.status_code == 201 - assert r.json["alias"] == f"prefix.{word}@{EMAIL_DOMAIN}" + assert r.json["alias"] == f"{prefix}.{word}@{EMAIL_DOMAIN}" # assert returned field res = r.json @@ -117,7 +120,7 @@ def test_full_payload(flask_client): assert new_alias.note == "test note" assert len(new_alias.mailboxes) == 2 - alias_used_on = AliasUsedOn.first() + alias_used_on = AliasUsedOn.filter(AliasUsedOn.user_id == user.id).first() assert alias_used_on.alias_id == new_alias.id assert alias_used_on.hostname == "example.com" @@ -250,6 +253,8 @@ def test_cannot_create_alias_in_trash(flask_client): def test_too_many_requests(flask_client): + config.DISABLE_RATE_LIMIT = False + user = login(flask_client) # create a custom domain diff --git a/tests/api/test_new_random_alias.py b/tests/api/test_new_random_alias.py index 22939ac0..8611d2ea 100644 --- a/tests/api/test_new_random_alias.py +++ b/tests/api/test_new_random_alias.py @@ -2,6 +2,7 @@ import uuid from flask import url_for, g +from app import config from app.config import EMAIL_DOMAIN, MAX_NB_EMAIL_FREE_PLAN from app.db import Session from app.models import Alias, CustomDomain, AliasUsedOn @@ -122,6 +123,7 @@ def test_out_of_quota(flask_client): def test_too_many_requests(flask_client): + config.DISABLE_RATE_LIMIT = False login(flask_client) # can't create more than 5 aliases in 1 minute diff --git a/tests/conftest.py b/tests/conftest.py index e6b92fc3..37b2abbd 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -42,15 +42,22 @@ def flask_app(): yield app +from app import config + + @pytest.fixture def flask_client(): transaction = connection.begin() with app.app_context(): + # disable rate limit during test + config.DISABLE_RATE_LIMIT = True try: client = app.test_client() yield client finally: + # disable rate limit again as some tests might enable rate limit + config.DISABLE_RATE_LIMIT = True # roll back all commits made during a test transaction.rollback() Session.rollback() diff --git a/tests/dashboard/test_api_keys.py b/tests/dashboard/test_api_keys.py index 7621b503..d2c36326 100644 --- a/tests/dashboard/test_api_keys.py +++ b/tests/dashboard/test_api_keys.py @@ -17,7 +17,7 @@ def test_api_key_page_requires_password(flask_client): def test_create_delete_api_key(flask_client): user = login(flask_client) - Session.commit() + nb_api_key = ApiKey.count() # to bypass sudo mode with flask_client.session_transaction() as session: @@ -31,7 +31,7 @@ def test_create_delete_api_key(flask_client): ) assert create_r.status_code == 200 api_key = ApiKey.get_by(user_id=user.id) - assert ApiKey.count() == 1 + assert ApiKey.filter(ApiKey.user_id == user.id).count() == 1 assert api_key.name == "for test" # delete api_key @@ -41,10 +41,12 @@ def test_create_delete_api_key(flask_client): follow_redirects=True, ) assert delete_r.status_code == 200 - assert ApiKey.count() == 0 + assert ApiKey.count() == nb_api_key def test_delete_all_api_keys(flask_client): + nb_api_keys = ApiKey.count() + # create two test users user_1 = login(flask_client) user_2 = User.create( @@ -59,7 +61,7 @@ def test_delete_all_api_keys(flask_client): Session.commit() assert ( - ApiKey.count() == 3 + ApiKey.count() == nb_api_keys + 3 ) # assert that the total number of API keys for all users is 3. # assert that each user has the API keys created assert ApiKey.filter(ApiKey.user_id == user_1.id).count() == 2 @@ -77,7 +79,7 @@ def test_delete_all_api_keys(flask_client): ) assert r.status_code == 200 assert ( - ApiKey.count() == 1 + ApiKey.count() == nb_api_keys + 1 ) # assert that the total number of API keys for all users is now 1. assert ( ApiKey.filter(ApiKey.user_id == user_1.id).count() == 0 diff --git a/tests/dashboard/test_custom_alias.py b/tests/dashboard/test_custom_alias.py index 72b4705a..31a4d690 100644 --- a/tests/dashboard/test_custom_alias.py +++ b/tests/dashboard/test_custom_alias.py @@ -2,6 +2,7 @@ from random import random from flask import url_for, g +from app import config from app.alias_suffix import ( get_alias_suffixes, AliasSuffix, @@ -328,6 +329,7 @@ def test_add_alias_in_custom_domain_trash(flask_client): def test_too_many_requests(flask_client): + config.DISABLE_RATE_LIMIT = False user = login(flask_client) # create a custom domain diff --git a/tests/dashboard/test_directory.py b/tests/dashboard/test_directory.py index eeb0a849..123970e2 100644 --- a/tests/dashboard/test_directory.py +++ b/tests/dashboard/test_directory.py @@ -62,10 +62,12 @@ def test_create_directory_in_trash(flask_client): def test_create_directory_out_of_quota(flask_client): user = login(flask_client) - for i in range(MAX_NB_DIRECTORY - Directory.count()): + for i in range( + MAX_NB_DIRECTORY - Directory.filter(Directory.user_id == user.id).count() + ): Directory.create(name=f"test{i}", user_id=user.id, commit=True) - assert Directory.count() == MAX_NB_DIRECTORY + assert Directory.filter(Directory.user_id == user.id).count() == MAX_NB_DIRECTORY flask_client.post( url_for("dashboard.directory"), @@ -74,4 +76,4 @@ def test_create_directory_out_of_quota(flask_client): ) # no new directory is created - assert Directory.count() == MAX_NB_DIRECTORY + assert Directory.filter(Directory.user_id == user.id).count() == MAX_NB_DIRECTORY diff --git a/tests/dashboard/test_index.py b/tests/dashboard/test_index.py index eff09c69..7f201569 100644 --- a/tests/dashboard/test_index.py +++ b/tests/dashboard/test_index.py @@ -1,5 +1,6 @@ from flask import url_for, g +from app import config from app.models import ( Alias, ) @@ -20,6 +21,7 @@ def test_create_random_alias_success(flask_client): def test_too_many_requests(flask_client): + config.DISABLE_RATE_LIMIT = False login(flask_client) # can't create more than 5 aliases in 1 minute diff --git a/tests/test_extensions.py b/tests/test_extensions.py index 673be5d0..a8f659f0 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -1,10 +1,12 @@ -from flask import g from http import HTTPStatus from random import Random +from flask import g + +from app import config from app.extensions import limiter -from tests.utils import login from tests.conftest import app as test_app +from tests.utils import login # IMPORTANT NOTICE # ---------------- @@ -41,6 +43,7 @@ def request_headers(source_ip: str) -> dict: def test_rate_limit_limits_by_source_ip(flask_client): + config.DISABLE_RATE_LIMIT = False source_ip = random_ip() for _ in range(_MAX_PER_MINUTE): @@ -59,6 +62,7 @@ def test_rate_limit_limits_by_source_ip(flask_client): def test_rate_limit_limits_by_user_id(flask_client): + config.DISABLE_RATE_LIMIT = False # Login with a user login(flask_client) fix_rate_limit_after_request() @@ -75,6 +79,7 @@ def test_rate_limit_limits_by_user_id(flask_client): def test_rate_limit_limits_by_user_id_ignoring_ip(flask_client): + config.DISABLE_RATE_LIMIT = False source_ip = random_ip() # Login with a user diff --git a/tests/test_models.py b/tests/test_models.py index 8fe4ea6a..01b3c4d1 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -18,7 +18,7 @@ from app.models import ( PlanEnum, PADDLE_SUBSCRIPTION_GRACE_DAYS, ) -from tests.utils import login, create_new_user +from tests.utils import login, create_new_user, random_token def test_generate_email(flask_client): @@ -71,45 +71,48 @@ def test_website_send_to(flask_client): alias = Alias.create_new_random(user) Session.commit() + prefix = random_token() + # non-empty name c1 = Contact.create( user_id=user.id, alias_id=alias.id, - website_email="abcd@example.com", + website_email=f"{prefix}@example.com", reply_email="rep@SL", name="First Last", ) - assert c1.website_send_to() == '"First Last | abcd at example.com" ' + assert c1.website_send_to() == f'"First Last | {prefix} at example.com" ' # empty name, ascii website_from, easy case c1.name = None - c1.website_from = "First Last " - assert c1.website_send_to() == '"First Last | abcd at example.com" ' + c1.website_from = f"First Last <{prefix}@example.com>" + assert c1.website_send_to() == f'"First Last | {prefix} at example.com" ' # empty name, RFC 2047 website_from c1.name = None - c1.website_from = "=?UTF-8?B?TmjGoW4gTmd1eeG7hW4=?= " - assert c1.website_send_to() == '"Nhơn Nguyễn | abcd at example.com" ' + c1.website_from = f"=?UTF-8?B?TmjGoW4gTmd1eeG7hW4=?= <{prefix}@example.com>" + assert c1.website_send_to() == f'"Nhơn Nguyễn | {prefix} at example.com" ' def test_new_addr_default_sender_format(flask_client): user = login(flask_client) alias = Alias.first() + prefix = random_token() contact = Contact.create( user_id=user.id, alias_id=alias.id, - website_email="abcd@example.com", + website_email=f"{prefix}@example.com", reply_email="rep@SL", name="First Last", commit=True, ) - assert contact.new_addr() == '"First Last - abcd at example.com" ' + assert contact.new_addr() == f'"First Last - {prefix} at example.com" ' # Make sure email isn't duplicated if sender name equals email - contact.name = "abcd@example.com" - assert contact.new_addr() == '"abcd at example.com" ' + contact.name = f"{prefix}@example.com" + assert contact.new_addr() == f'"{prefix} at example.com" ' def test_new_addr_a_sender_format(flask_client): @@ -117,17 +120,18 @@ def test_new_addr_a_sender_format(flask_client): user.sender_format = SenderFormatEnum.A.value Session.commit() alias = Alias.first() + prefix = random_token() contact = Contact.create( user_id=user.id, alias_id=alias.id, - website_email="abcd@example.com", + website_email=f"{prefix}@example.com", reply_email="rep@SL", name="First Last", commit=True, ) - assert contact.new_addr() == '"First Last - abcd(a)example.com" ' + assert contact.new_addr() == f'"First Last - {prefix}(a)example.com" ' def test_new_addr_no_name_sender_format(flask_client): @@ -135,11 +139,12 @@ def test_new_addr_no_name_sender_format(flask_client): user.sender_format = SenderFormatEnum.NO_NAME.value Session.commit() alias = Alias.first() + prefix = random_token() contact = Contact.create( user_id=user.id, alias_id=alias.id, - website_email="abcd@example.com", + website_email=f"{prefix}@example.com", reply_email="rep@SL", name="First Last", commit=True, @@ -153,11 +158,12 @@ def test_new_addr_name_only_sender_format(flask_client): user.sender_format = SenderFormatEnum.NAME_ONLY.value Session.commit() alias = Alias.first() + prefix = random_token() contact = Contact.create( user_id=user.id, alias_id=alias.id, - website_email="abcd@example.com", + website_email=f"{prefix}@example.com", reply_email="rep@SL", name="First Last", commit=True, @@ -171,27 +177,29 @@ def test_new_addr_at_only_sender_format(flask_client): user.sender_format = SenderFormatEnum.AT_ONLY.value Session.commit() alias = Alias.first() + prefix = random_token() contact = Contact.create( user_id=user.id, alias_id=alias.id, - website_email="abcd@example.com", + website_email=f"{prefix}@example.com", reply_email="rep@SL", name="First Last", commit=True, ) - assert contact.new_addr() == '"abcd at example.com" ' + assert contact.new_addr() == f'"{prefix} at example.com" ' def test_new_addr_unicode(flask_client): user = login(flask_client) alias = Alias.first() + random_prefix = random_token() contact = Contact.create( user_id=user.id, alias_id=alias.id, - website_email="abcd@example.com", + website_email=f"{random_prefix}@example.com", reply_email="rep@SL", name="Nhơn Nguyễn", commit=True, @@ -199,12 +207,12 @@ def test_new_addr_unicode(flask_client): assert ( contact.new_addr() - == "=?utf-8?q?Nh=C6=A1n_Nguy=E1=BB=85n_-_abcd_at_example=2Ecom?= " + == f"=?utf-8?q?Nh=C6=A1n_Nguy=E1=BB=85n_-_{random_prefix}_at_example=2Ecom?= " ) # sanity check assert parse_full_address(contact.new_addr()) == ( - "Nhơn Nguyễn - abcd at example.com", + f"Nhơn Nguyễn - {random_prefix} at example.com", "rep@sl", )