Skip to content

Commit 297eb8d

Browse files
tozotozo
committed
fixing the findings on the PR:
making sure the query and delete functionality is specific to a given user additional checks to data that's coming from the front-end
1 parent 8bcc97b commit 297eb8d

7 files changed

Lines changed: 216 additions & 5 deletions

File tree

‎app/custom_domain_utils.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -135,7 +135,7 @@ def can_blocked_domain_be_used(
135135
if reason is not None:
136136
return reason
137137

138-
if BlockedDomain.get_by(domain=domain):
138+
if BlockedDomain.get_by(user_id=user.id, domain=domain):
139139
return CannotUseDomainReason.DomainAlreadyUsed
140140
else:
141141
return None

‎app/dashboard/views/setting.py‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -294,6 +294,11 @@ def setting():
294294
flash("Your preference has been updated", "success")
295295
elif request.form.get("form-name") == "blocked-domains-add":
296296
domain = request.form.get("domain-name")
297+
298+
if not domain:
299+
flash("Domain name is required", "error")
300+
return redirect(url_for("dashboard.setting"))
301+
297302
new_domain = sanitize_domain(domain)
298303
domain_forbidden_cause = can_blocked_domain_be_used(
299304
current_user, new_domain
@@ -305,17 +310,42 @@ def setting():
305310

306311
BlockedDomain.create(user_id=current_user.id, domain=new_domain)
307312

313+
LOG.i(
314+
f"A new blocked domain [{new_domain}] was added for user [{current_user.id}]"
315+
)
316+
308317
Session.commit()
309318
flash(f"Added blocked domain [{domain}]", "success")
310319
return redirect(url_for("dashboard.setting"))
311320
elif request.form.get("form-name") == "blocked-domains-remove":
312321
domain_id = request.form.get("domain_id")
313322
domain_name = request.form.get("domain_name")
314323

324+
if not domain_id:
325+
flash("Domain Id is missing", "error")
326+
return redirect(url_for("dashboard.setting"))
327+
328+
if not domain_name:
329+
flash("Domain Name is missing", "error")
330+
return redirect(url_for("dashboard.setting"))
331+
315332
domain = BlockedDomain.get(domain_id)
333+
if not domain or domain.user_id != current_user.id:
334+
LOG.e(
335+
f"Blocked domain with id [{domain_id}] not found or not owned by user [{current_user.id}] therefore couldn't be deleted"
336+
)
337+
flash(
338+
"Blocked domain not found or the user doesn't have access to it",
339+
"error",
340+
)
341+
return redirect(url_for("dashboard.setting"))
316342

317343
BlockedDomain.delete(domain.id)
318344

345+
LOG.i(
346+
f"A blocked domain [{domain}] was deleted by user [{current_user.id}]"
347+
)
348+
319349
Session.commit()
320350
flash(f"Deleted blocked domain [{domain_name}]", "success")
321351
return redirect(url_for("dashboard.setting"))

‎app/models.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -503,7 +503,7 @@ class User(Base, ModelMixin, UserMixin, PasswordOracle):
503503

504504
default_mailbox = orm.relationship("Mailbox", foreign_keys=[default_mailbox_id])
505505

506-
_blocked_domains = orm.relationship("BlockedDomain", lazy="joined")
506+
_blocked_domains = orm.relationship("BlockedDomain")
507507

508508
# user can set a more strict max_spam score to block spams more aggressively
509509
max_spam_score = sa.Column(sa.Integer, nullable=True)

‎email_handler.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -615,9 +615,15 @@ def handle_forward(envelope, msg: Message, rcpt_to: str) -> List[Tuple[bool, str
615615

616616
mail_from_domain = get_email_domain_part(mail_from)
617617
if is_domain_blocked(user.id, mail_from_domain):
618+
LOG.i(
619+
f"Email [{mail_from}] was ignored for the user [{user.id}] because of a blocked domain [{mail_from_domain}]"
620+
)
618621
# by default return 2** instead of 5** to allow user to receive emails again when domain is unblocked
619622
res_status = status.E200
620623
if user.block_behaviour == BlockBehaviourEnum.return_5xx:
624+
LOG.i(
625+
f"Email [{mail_from}] was rejected for the user [{user.id}] because of a blocked domain [{mail_from_domain}]"
626+
)
621627
res_status = status.E502
622628

623629
return [(True, res_status)]

‎migrations/versions/2025_101020_fdb1b61d9bdb_.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
"""empty message
22
33
Revision ID: c18048c40ed9
4-
Revises: 3ffdea52697d
4+
Revises: 3ee37864eb67
55
Create Date: 2025-10-10 20:29:32.701784
66
77
"""
@@ -11,7 +11,7 @@
1111

1212
# revision identifiers, used by Alembic.
1313
revision = 'c18048c40ed9'
14-
down_revision = '3ffdea52697d'
14+
down_revision = '3ee37864eb67'
1515
branch_labels = None
1616
depends_on = None
1717

‎tests/dashboard/test_setting.py‎

Lines changed: 160 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,10 @@
1+
from unittest.mock import patch
2+
13
from flask import url_for
24

35
from app import config
4-
from app.models import EmailChange
6+
from app.db import Session
7+
from app.models import EmailChange, BlockedDomain
58
from app.utils import canonicalize_email
69
from tests.utils import login, random_email, create_new_user
710

@@ -26,3 +29,159 @@ def test_setup_done(flask_client):
2629
assert email_change is not None
2730
assert email_change.new_email == canonicalize_email(noncanonical_email)
2831
config.SKIP_MX_LOOKUP_ON_CHECK = False
32+
33+
34+
def test_add_blocked_domain_none(flask_client):
35+
user = create_new_user()
36+
login(flask_client, user)
37+
38+
# Missing domain-name
39+
r = flask_client.post(
40+
url_for("dashboard.setting"),
41+
data={
42+
"form-name": "blocked-domains-add",
43+
},
44+
follow_redirects=True,
45+
)
46+
assert r.status_code == 200
47+
assert BlockedDomain.filter_by(user_id=user.id).count() == 0
48+
49+
50+
def test_add_blocked_domain_empty(flask_client):
51+
user = create_new_user()
52+
login(flask_client, user)
53+
54+
# Empty domain-name
55+
r = flask_client.post(
56+
url_for("dashboard.setting"),
57+
data={
58+
"form-name": "blocked-domains-add",
59+
"domain-name": "",
60+
},
61+
follow_redirects=True,
62+
)
63+
assert r.status_code == 200
64+
assert BlockedDomain.filter_by(user_id=user.id).count() == 0
65+
66+
67+
def test_add_blocked_domain_success(flask_client):
68+
user = create_new_user()
69+
login(flask_client, user)
70+
71+
r = flask_client.post(
72+
url_for("dashboard.setting"),
73+
data={
74+
"form-name": "blocked-domains-add",
75+
"domain-name": " ExAmple.COM ",
76+
},
77+
follow_redirects=True,
78+
)
79+
assert r.status_code == 200
80+
81+
blocked_domains = BlockedDomain.filter_by(user_id=user.id).all()
82+
assert len(blocked_domains) == 1
83+
assert blocked_domains[0].domain == "example.com"
84+
85+
86+
def test_remove_blocked_domain_no_id(flask_client):
87+
user = create_new_user()
88+
89+
login(flask_client, user)
90+
91+
# Missing domain_id
92+
with patch("app.dashboard.views.setting.BlockedDomain.delete") as mock_delete:
93+
r = flask_client.post(
94+
url_for("dashboard.setting"),
95+
data={
96+
"form-name": "blocked-domains-remove",
97+
"domain_name": "example.com",
98+
},
99+
follow_redirects=True,
100+
)
101+
assert r.status_code == 200
102+
mock_delete.assert_not_called()
103+
104+
105+
def test_remove_blocked_domain_invalid_id(flask_client):
106+
user = create_new_user()
107+
login(flask_client, user)
108+
109+
# Invalid domain_id
110+
with patch("app.dashboard.views.setting.BlockedDomain.delete") as mock_delete:
111+
r = flask_client.post(
112+
url_for("dashboard.setting"),
113+
data={
114+
"form-name": "blocked-domains-remove",
115+
"domain_id": 9999,
116+
"domain_name": "example.com",
117+
},
118+
follow_redirects=True,
119+
)
120+
assert r.status_code == 200
121+
mock_delete.assert_not_called()
122+
123+
124+
def test_remove_blocked_domain_success(flask_client):
125+
user = create_new_user()
126+
bd = BlockedDomain.create(user_id=user.id, domain="example.com")
127+
Session.commit()
128+
129+
login(flask_client, user)
130+
131+
r = flask_client.post(
132+
url_for("dashboard.setting"),
133+
data={
134+
"form-name": "blocked-domains-remove",
135+
"domain_id": bd.id,
136+
"domain_name": bd.domain,
137+
},
138+
follow_redirects=True,
139+
)
140+
assert r.status_code == 200
141+
assert BlockedDomain.get(bd.id) is None
142+
143+
144+
def test_remove_blocked_domain_not_owned(flask_client):
145+
user1 = create_new_user()
146+
user2 = create_new_user()
147+
148+
bd = BlockedDomain.create(user_id=user1.id, domain="example.com")
149+
Session.commit()
150+
151+
login(flask_client, user2)
152+
153+
# Try to remove user1's blocked domain as user2
154+
r = flask_client.post(
155+
url_for("dashboard.setting"),
156+
data={
157+
"form-name": "blocked-domains-remove",
158+
"domain_id": bd.id,
159+
"domain_name": "example.com",
160+
},
161+
follow_redirects=True,
162+
)
163+
assert r.status_code == 200
164+
assert BlockedDomain.get(bd.id) is not None
165+
166+
167+
def test_remove_blocked_domain_name_missing(flask_client):
168+
user = create_new_user()
169+
bd = BlockedDomain.create(user_id=user.id, domain="example.com")
170+
Session.commit()
171+
172+
login(flask_client, user)
173+
174+
# Missing domain_name
175+
with patch("app.dashboard.views.setting.BlockedDomain.delete") as mock_delete:
176+
r = flask_client.post(
177+
url_for("dashboard.setting"),
178+
data={
179+
"form-name": "blocked-domains-remove",
180+
"domain_id": bd.id,
181+
},
182+
follow_redirects=True,
183+
)
184+
assert r.status_code == 200
185+
# Check that it didn't delete the domain
186+
assert BlockedDomain.get(bd.id) is not None
187+
mock_delete.assert_not_called()

‎tests/test_custom_domain_utils.py‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -161,6 +161,22 @@ def test_can_blocked_domain_be_used_existing_domain():
161161
assert res is CannotUseDomainReason.DomainAlreadyUsed
162162

163163

164+
def test_can_blocked_domain_be_used_different_users():
165+
domain = random_domain()
166+
other_user = create_new_user()
167+
168+
# User 1 blocks the domain
169+
BlockedDomain.create(user_id=user.id, domain=domain, commit=True)
170+
171+
# User 1 should not be able to block it again
172+
res1 = can_blocked_domain_be_used(user, domain)
173+
assert res1 is CannotUseDomainReason.DomainAlreadyUsed
174+
175+
# User 2 should be able to block it
176+
res2 = can_blocked_domain_be_used(other_user, domain)
177+
assert res2 is None
178+
179+
164180
# sanitize_domain
165181
def test_can_sanitize_domain_empty():
166182
assert sanitize_domain("") == ""

0 commit comments

Comments
 (0)