Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Move typo checks to after_insert #17764

Draft
wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Move typo checks to after_insert #17764

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 89 additions & 0 deletions warehouse/packaging/models.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
Text,
UniqueConstraint,
cast,
event,
func,
or_,
orm,
Expand Down Expand Up @@ -67,6 +68,7 @@
from warehouse.classifiers.models import Classifier
from warehouse.events.models import HasEvents
from warehouse.forklift import metadata
from warehouse.helpdesk.interfaces import IAdminNotificationService
from warehouse.integrations.vulnerabilities.models import VulnerabilityRecord
from warehouse.observations.models import HasObservations
from warehouse.organizations.models import (
Expand All @@ -77,6 +79,10 @@
Team,
TeamProjectRole,
)
from warehouse.packaging.interfaces import (
IProjectService,
ProjectNameUnavailableTypoSquattingError,
)
from warehouse.sitemap.models import SitemapMixin
from warehouse.utils import dotted_navigator, wheel
from warehouse.utils.attrs import make_repr
Expand Down Expand Up @@ -496,6 +502,89 @@ def yanked_releases(self):
)


@event.listens_for(Project, "after_insert")
def receive_after_insert(mapper, connection, project):
request = get_current_request()
project_service = request.find_service(IProjectService)
name = project.name
try:
project_service.check_project_name_after_insert(name)
except ProjectNameUnavailableTypoSquattingError as exc:
request.log.warning(
"ProjectNameUnavailableTypoSquattingError",
check_name=exc.check_name,
existing_project_name=exc.existing_project_name,
)
# Send notification to Admins for review
notification_service = request.find_service(IAdminNotificationService)

warehouse_domain = request.registry.settings.get("warehouse.domain")
new_project_page = request.route_url(
"packaging.project",
name=name,
_host=warehouse_domain,
)
new_project_text = (
f"During `file_upload`, Project Create for "
f"*<{new_project_page}|{name}>* was detected as a potential "
f"typo by the `{exc.check_name!r}` check."
)
existing_project_page = request.route_url(
"packaging.project",
name=exc.existing_project_name,
_host=warehouse_domain,
)
existing_project_text = (
f"<{existing_project_page}|Existing project: "
f"{exc.existing_project_name}>"
)

webhook_payload = {
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": "TypoSnyper :warning:",
"emoji": True,
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": new_project_text,
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": existing_project_text,
},
},
{"type": "divider"},
{
"type": "context",
"elements": [
{
"type": "plain_text",
"text": "Once reviewed/confirmed, "
"react to this message with :white_check_mark:",
"emoji": True,
}
],
},
]
}
notification_service.send_notification(payload=webhook_payload)

request.metrics.increment(
"warehouse.packaging.services.create_project.typo_squatting",
tags=[f"check_name:{exc.check_name!r}"],
)


class DependencyKind(enum.IntEnum):
requires = 1
provides = 2
Expand Down
83 changes: 3 additions & 80 deletions warehouse/packaging/services.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,6 @@
from warehouse.admin.flags import AdminFlagValue
from warehouse.email import send_pending_trusted_publisher_invalidated_email
from warehouse.events.tags import EventTag
from warehouse.helpdesk.interfaces import IAdminNotificationService
from warehouse.metrics import IMetricsService
from warehouse.oidc.models import PendingOIDCPublisher
from warehouse.packaging.interfaces import (
Expand Down Expand Up @@ -485,6 +484,9 @@ def check_project_name(self, name: str) -> None:
).first():
raise ProjectNameUnavailableSimilarError(similar_project_name)

return None

def check_project_name_after_insert(self, name: str) -> None:
# Check for typo-squatting.
if typo_check_match := typo_check_name(canonicalize_name(name)):
raise ProjectNameUnavailableTypoSquattingError(
Expand Down Expand Up @@ -558,85 +560,6 @@ def create_project(
projecthelp=request.help_url(_anchor="project-name"),
),
) from None
except ProjectNameUnavailableTypoSquattingError as exc:
# Don't yet raise an error here, as we want to allow the
# user to proceed with the project creation. We'll log a warning
# instead.
request.log.warning(
"ProjectNameUnavailableTypoSquattingError",
check_name=exc.check_name,
existing_project_name=exc.existing_project_name,
)
# Send notification to Admins for review
notification_service = request.find_service(IAdminNotificationService)

warehouse_domain = request.registry.settings.get("warehouse.domain")
new_project_page = request.route_url(
"packaging.project",
name=name,
_host=warehouse_domain,
)
new_project_text = (
f"During `file_upload`, Project Create for "
f"*<{new_project_page}|{name}>* was detected as a potential "
f"typo by the `{exc.check_name!r}` check."
)
existing_project_page = request.route_url(
"packaging.project",
name=exc.existing_project_name,
_host=warehouse_domain,
)
existing_project_text = (
f"<{existing_project_page}|Existing project: "
f"{exc.existing_project_name}>"
)

webhook_payload = {
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": "TypoSnyper :warning:",
"emoji": True,
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": new_project_text,
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": existing_project_text,
},
},
{"type": "divider"},
{
"type": "context",
"elements": [
{
"type": "plain_text",
"text": "Once reviewed/confirmed, "
"react to this message with :white_check_mark:",
"emoji": True,
}
],
},
]
}
notification_service.send_notification(payload=webhook_payload)

request.metrics.increment(
"warehouse.packaging.services.create_project.typo_squatting",
tags=[f"check_name:{exc.check_name!r}"],
)
# and continue with the project creation
pass

# The project name is valid: create it and add it
project = Project(name=name)
Expand Down
Loading