# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import collections
import datetime
import re

import elasticsearch

from pyramid.exceptions import PredicateMismatch
from pyramid.httpexceptions import (
    HTTPBadRequest,
    HTTPException,
    HTTPMovedPermanently,
    HTTPNotFound,
    HTTPSeeOther,
    HTTPServiceUnavailable,
    exception_response,
)
from pyramid.i18n import make_localizer
from pyramid.interfaces import ITranslationDirectories
from pyramid.renderers import render_to_response
from pyramid.response import Response
from pyramid.view import (
    exception_view_config,
    forbidden_view_config,
    notfound_view_config,
    view_config,
    view_defaults,
)
from sqlalchemy import func
from sqlalchemy.orm import aliased, joinedload
from sqlalchemy.orm.exc import NoResultFound
from sqlalchemy.sql import exists, expression
from trove_classifiers import deprecated_classifiers, sorted_classifiers

from warehouse.accounts import REDIRECT_FIELD_NAME
from warehouse.accounts.forms import TitanPromoCodeForm
from warehouse.accounts.models import TitanPromoCode, User
from warehouse.cache.http import add_vary, cache_control
from warehouse.cache.origin import origin_cache
from warehouse.classifiers.models import Classifier
from warehouse.db import DatabaseNotAvailableError
from warehouse.errors import WarehouseDenied
from warehouse.forms import SetLocaleForm
from warehouse.i18n import LOCALE_ATTR
from warehouse.manage.views import user_projects
from warehouse.metrics import IMetricsService
from warehouse.packaging.models import File, Project, Release, release_classifiers
from warehouse.search.queries import SEARCH_FILTER_ORDER, get_es_query
from warehouse.utils.http import is_safe_url
from warehouse.utils.paginate import ElasticsearchPage, paginate_url_factory
from warehouse.utils.row_counter import RowCount

JSON_REGEX = r"^/pypi/([^\/]+)\/?([^\/]+)?/json\/?$"
json_path = re.compile(JSON_REGEX)


@view_config(context=HTTPException)
@notfound_view_config(append_slash=HTTPMovedPermanently)
@notfound_view_config(path_info=JSON_REGEX, append_slash=False)
def httpexception_view(exc, request):
    # This special case exists for the easter egg that appears on the 404
    # response page. We don't generally allow youtube embeds, but we make an
    # except for this one.
    if isinstance(exc, HTTPNotFound):
        request.find_service(name="csp").merge(
            {
                "frame-src": ["https://www.youtube-nocookie.com"],
                "script-src": ["https://www.youtube.com", "https://s.ytimg.com"],
            }
        )
    try:
        # Lightweight version of 404 page for `/simple/`
        if isinstance(exc, HTTPNotFound) and request.path.startswith("/simple/"):
            response = Response(body="404 Not Found", content_type="text/plain")
        elif isinstance(exc, HTTPNotFound) and json_path.match(request.path):
            response = Response(
                body='{"message": "Not Found"}',
                charset="utf-8",
                content_type="application/json",
            )
        else:
            response = render_to_response(
                "{}.html".format(exc.status_code), {}, request=request
            )
    except LookupError:
        # We don't have a customized template for this error, so we'll just let
        # the default happen instead.
        return exc

    # Copy over the important values from our HTTPException to our new response
    # object.
    response.status = exc.status
    response.headers.extend(
        (k, v) for k, v in exc.headers.items() if k not in response.headers
    )

    return response


@forbidden_view_config()
@exception_view_config(PredicateMismatch)
def forbidden(exc, request):
    # If the forbidden error is because the user isn't logged in, then we'll
    # redirect them to the log in page.
    if request.authenticated_userid is None:
        url = request.route_url(
            "accounts.login", _query={REDIRECT_FIELD_NAME: request.path_qs}
        )
        return HTTPSeeOther(url)

    # Check if the error has a "result" attribute and if it is a WarehouseDenied
    if hasattr(exc, "result") and isinstance(exc.result, WarehouseDenied):
        # If the forbidden error is because the user doesn't have 2FA enabled, we'll
        # redirect them to the 2FA page
        if exc.result.reason in {"owners_require_2fa", "pypi_mandates_2fa"}:
            request.session.flash(
                request._(
                    "Two-factor authentication must be enabled on your account to "
                    "perform this action."
                ),
                queue="error",
            )
            url = request.route_url(
                "manage.account.two-factor",
                _query={REDIRECT_FIELD_NAME: request.path_qs},
            )
            return HTTPSeeOther(url)

    # If we've reached here, then the user is logged in and they are genuinely
    # not allowed to access this page.
    return httpexception_view(exc, request)


@forbidden_view_config(path_info=r"^/_includes/")
@exception_view_config(PredicateMismatch, path_info=r"^/_includes/")
def forbidden_include(exc, request):
    # If the forbidden error is for a client-side-include, just return an empty
    # response instead of redirecting
    return Response(status=403)


@view_config(context=DatabaseNotAvailableError)
def service_unavailable(exc, request):
    return httpexception_view(HTTPServiceUnavailable(), request)


@view_config(
    route_name="robots.txt",
    renderer="robots.txt",
    decorator=[
        cache_control(1 * 24 * 60 * 60),  # 1 day
        origin_cache(
            1 * 24 * 60 * 60,  # 1 day
            stale_while_revalidate=6 * 60 * 60,  # 6 hours
            stale_if_error=1 * 24 * 60 * 60,  # 1 day
        ),
    ],
)
def robotstxt(request):
    request.response.content_type = "text/plain"
    return {}


@view_config(
    route_name="opensearch.xml",
    renderer="opensearch.xml",
    decorator=[
        cache_control(1 * 24 * 60 * 60),  # 1 day
        origin_cache(
            1 * 24 * 60 * 60,  # 1 day
            stale_while_revalidate=6 * 60 * 60,  # 6 hours
            stale_if_error=1 * 24 * 60 * 60,  # 1 day
        ),
    ],
)
def opensearchxml(request):
    request.response.content_type = "text/xml"
    return {}


@view_config(
    route_name="index",
    renderer="index.html",
    decorator=[
        origin_cache(
            1 * 60 * 60,  # 1 hour
            stale_while_revalidate=10 * 60,  # 10 minutes
            stale_if_error=1 * 24 * 60 * 60,  # 1 day
            keys=["all-projects", "trending"],
        )
    ],
    has_translations=True,
)
def index(request):
    project_ids = [
        r[0]
        for r in (
            request.db.query(Project.id)
            .order_by(Project.zscore.desc().nullslast(), func.random())
            .limit(5)
            .all()
        )
    ]
    release_a = aliased(
        Release,
        request.db.query(Release)
        .distinct(Release.project_id)
        .filter(Release.project_id.in_(project_ids))
        .order_by(
            Release.project_id,
            Release.is_prerelease.nullslast(),
            Release._pypi_ordering.desc(),
        )
        .subquery(),
    )
    trending_projects = (
        request.db.query(release_a)
        .options(joinedload(release_a.project))
        .order_by(func.array_idx(project_ids, release_a.project_id))
        .all()
    )

    latest_releases = (
        request.db.query(Release)
        .options(joinedload(Release.project))
        .order_by(Release.created.desc())
        .limit(5)
        .all()
    )

    counts = dict(
        request.db.query(RowCount.table_name, RowCount.count)
        .filter(
            RowCount.table_name.in_(
                [
                    Project.__tablename__,
                    Release.__tablename__,
                    File.__tablename__,
                    User.__tablename__,
                ]
            )
        )
        .all()
    )

    return {
        "latest_releases": latest_releases,
        "trending_projects": trending_projects,
        "num_projects": counts.get(Project.__tablename__, 0),
        "num_releases": counts.get(Release.__tablename__, 0),
        "num_files": counts.get(File.__tablename__, 0),
        "num_users": counts.get(User.__tablename__, 0),
    }


@view_config(
    route_name="locale",
    request_method="GET",
    request_param=SetLocaleForm.__params__,
    uses_session=True,
)
def locale(request):
    try:
        form = SetLocaleForm(locale_id=request.GET.getone("locale_id"))
    except KeyError:
        raise HTTPBadRequest("Invalid amount of locale_id parameters provided")

    redirect_to = request.referer
    if not is_safe_url(redirect_to, host=request.host):
        redirect_to = request.route_path("index")
    resp = HTTPSeeOther(redirect_to)

    if form.validate():
        # Build a localizer for the locale we're about to switch to. This will
        # happen automatically once the cookie is set, but if we want the flash
        # message indicating success to be in the new language as well, we need
        # to do it here.
        tdirs = request.registry.queryUtility(ITranslationDirectories)
        _ = make_localizer(form.locale_id.data, tdirs).translate
        request.session.flash(_("Locale updated"), queue="success")
        resp.set_cookie(LOCALE_ATTR, form.locale_id.data)

    return resp


@view_config(
    route_name="classifiers", renderer="pages/classifiers.html", has_translations=True
)
def list_classifiers(request):
    return {"classifiers": sorted_classifiers}


@view_config(
    route_name="search",
    renderer="search/results.html",
    decorator=[
        origin_cache(
            1 * 60 * 60,  # 1 hour
            stale_if_error=1 * 24 * 60 * 60,  # 1 day
            keys=["all-projects"],
        )
    ],
    has_translations=True,
)
def search(request):
    metrics = request.find_service(IMetricsService, context=None)

    querystring = request.params.get("q", "").replace("'", '"')
    order = request.params.get("o", "")
    classifiers = request.params.getall("c")
    query = get_es_query(request.es, querystring, order, classifiers)

    try:
        page_num = int(request.params.get("page", 1))
    except ValueError:
        raise HTTPBadRequest("'page' must be an integer.")

    try:
        page = ElasticsearchPage(
            query, page=page_num, url_maker=paginate_url_factory(request)
        )
    except elasticsearch.TransportError:
        metrics.increment("warehouse.views.search.error")
        raise HTTPServiceUnavailable

    if page.page_count and page_num > page.page_count:
        raise HTTPNotFound

    available_filters = collections.defaultdict(list)

    classifiers_q = (
        request.db.query(Classifier)
        .with_entities(Classifier.classifier)
        .filter(
            exists([release_classifiers.c.trove_id]).where(
                release_classifiers.c.trove_id == Classifier.id
            ),
            Classifier.classifier.notin_(deprecated_classifiers.keys()),
        )
        .order_by(
            expression.case(
                {c: i for i, c in enumerate(sorted_classifiers)},
                value=Classifier.classifier,
            )
        )
    )

    for cls in classifiers_q:
        first, *_ = cls.classifier.split(" :: ")
        available_filters[first].append(cls.classifier)

    def filter_key(item):
        try:
            return 0, SEARCH_FILTER_ORDER.index(item[0]), item[0]
        except ValueError:
            return 1, 0, item[0]

    def form_filters_tree(split_list):
        """
        Takes a list of lists, each of them containing a filter and
        one of its children.
        Returns a dictionary, each key being a filter and each value being
        the filter's children.
        """
        d = {}
        for list_ in split_list:
            current_level = d
            for part in list_:
                if part not in current_level:
                    current_level[part] = {}
                current_level = current_level[part]
        return d

    def process_available_filters():
        """
        Processes available filters and returns a list of dictionaries.
        The value of a key in the dictionary represents its children
        """
        sorted_filters = sorted(available_filters.items(), key=filter_key)
        output = []
        for f in sorted_filters:
            classifier_list = f[1]
            split_list = [i.split(" :: ") for i in classifier_list]
            tree = form_filters_tree(split_list)
            output.append(tree)
        return output

    metrics = request.find_service(IMetricsService, context=None)
    metrics.histogram("warehouse.views.search.results", page.item_count)

    return {
        "page": page,
        "term": querystring,
        "order": order,
        "available_filters": process_available_filters(),
        "applied_filters": request.params.getall("c"),
    }


@view_config(
    route_name="stats",
    renderer="pages/stats.html",
    decorator=[
        add_vary("Accept"),
        cache_control(1 * 24 * 60 * 60),  # 1 day
        origin_cache(
            1 * 24 * 60 * 60,  # 1 day
            stale_while_revalidate=1 * 24 * 60 * 60,  # 1 day
            stale_if_error=1 * 24 * 60 * 60,  # 1 day
        ),
    ],
    has_translations=True,
)
@view_config(
    route_name="stats.json",
    renderer="json",
    decorator=[
        add_vary("Accept"),
        cache_control(1 * 24 * 60 * 60),  # 1 day
        origin_cache(
            1 * 24 * 60 * 60,  # 1 day
            stale_while_revalidate=1 * 24 * 60 * 60,  # 1 day
            stale_if_error=1 * 24 * 60 * 60,  # 1 day
        ),
    ],
    accept="application/json",
)
def stats(request):
    total_size = int(request.db.query(func.sum(Project.total_size)).first()[0])
    top_100_packages = (
        request.db.query(Project)
        .with_entities(Project.name, Project.total_size)
        .order_by(Project.total_size.desc().nullslast())
        .limit(100)
        .all()
    )
    # Move top packages into a dict to make JSON more self describing
    top_packages = {
        pkg_name: {"size": int(pkg_bytes) if pkg_bytes is not None else 0}
        for pkg_name, pkg_bytes in top_100_packages
    }

    return {"total_packages_size": total_size, "top_packages": top_packages}


@view_defaults(
    route_name="security-key-giveaway",
    renderer="pages/security-key-giveaway.html",
    uses_session=True,
    has_translations=True,
    require_csrf=True,
    require_methods=False,
)
class SecurityKeyGiveaway:
    def __init__(self, request):
        self.request = request

    @property
    def form(self):
        return TitanPromoCodeForm(**self.request.POST)

    @property
    def codes_available(self):
        return (
            self.request.db.query(TitanPromoCode)
            .filter(TitanPromoCode.user_id.is_(None))
            .count()
        ) > 0

    @property
    def eligible_projects(self):
        if self.request.user:
            return set(
                project.name
                for project in user_projects(self.request)["projects_requiring_2fa"]
            )
        else:
            return set()

    @property
    def promo_code(self):
        if self.request.user:
            try:
                return (
                    self.request.db.query(TitanPromoCode).filter(
                        TitanPromoCode.user_id == self.request.user.id
                    )
                ).one()
            except NoResultFound:
                pass
        return None

    @property
    def default_response(self):
        codes_available = self.codes_available
        eligible_projects = self.eligible_projects
        promo_code = self.promo_code
        has_webauthn = self.request.user and self.request.user.has_webauthn

        eligible = (
            codes_available
            and bool(eligible_projects)
            and not has_webauthn
            and not promo_code
        )

        if not codes_available:
            reason_ineligible = "At this time there are no keys available"
        elif not eligible_projects:
            reason_ineligible = "You are not a collaborator on any critical projects"
        elif has_webauthn:
            reason_ineligible = (
                "You already have two-factor authentication enabled with a hardware "
                "security key"
            )
        elif promo_code:
            reason_ineligible = "Promo code has already been generated"
        else:
            reason_ineligible = None

        return {
            "eligible": eligible,
            "reason_ineligible": reason_ineligible,
            "form": self.form,
            "codes_available": self.codes_available,
            "eligible_projects": self.eligible_projects,
            "promo_code": self.promo_code,
            "REDIRECT_FIELD_NAME": REDIRECT_FIELD_NAME,
        }

    @view_config(request_method="GET")
    def security_key_giveaway(self):
        if not self.request.registry.settings.get(
            "warehouse.two_factor_mandate.available"
        ):
            raise HTTPNotFound

        return self.default_response

    @view_config(request_method="POST")
    def security_key_giveaway_submit(self):
        if not self.request.registry.settings.get(
            "warehouse.two_factor_mandate.available"
        ):
            raise HTTPNotFound

        default_response = self.default_response

        if not self.form.validate():
            self.request.session.flash("Form is not valid")
        elif not default_response["eligible"]:
            self.request.session.flash(default_response["reason_ineligible"])
        else:
            # The form is valid, assign a promo code to the user
            promo_code = (
                self.request.db.query(TitanPromoCode).filter(
                    TitanPromoCode.user_id.is_(None)
                )
            ).first()
            promo_code.user_id = self.request.user.id
            promo_code.distributed = datetime.datetime.now()
            # Flush so the promo code is available for the response
            self.request.db.flush()
            default_response["promo_code"] = promo_code
        return default_response


@view_config(
    route_name="includes.current-user-indicator",
    renderer="includes/current-user-indicator.html",
    uses_session=True,
    has_translations=True,
)
def current_user_indicator(request):
    return {}


@view_config(
    route_name="includes.flash-messages",
    renderer="includes/flash-messages.html",
    uses_session=True,
    has_translations=True,
)
def flash_messages(request):
    return {}


@view_config(
    route_name="includes.session-notifications",
    renderer="includes/session-notifications.html",
    uses_session=True,
    has_translations=True,
)
def session_notifications(request):
    return {}


@view_config(
    route_name="includes.sidebar-sponsor-logo",
    renderer="includes/sidebar-sponsor-logo.html",
    uses_session=False,
    has_translations=False,
    decorator=[
        cache_control(30),  # 30 seconds
    ],
)
def sidebar_sponsor_logo(request):
    return {}


@view_config(route_name="health", renderer="string")
def health(request):
    # This will ensure that we can access the database and run queries against
    # it without doing anything that will take a lock or block other queries.
    request.db.execute("SELECT 1")

    # Nothing will actually check this, but it's a little nicer to have
    # something to return besides an empty body.
    return "OK"


@view_config(route_name="force-status")
def force_status(request):
    try:
        raise exception_response(int(request.matchdict["status"]))
    except KeyError:
        raise exception_response(404) from None

warehouse/views.py
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import collections
import datetime
import re

import elasticsearch

from pyramid.exceptions import PredicateMismatch
from pyramid.httpexceptions import (
    HTTPBadRequest,
    HTTPException,
    HTTPMovedPermanently,
    HTTPNotFound,
    HTTPSeeOther,
    HTTPServiceUnavailable,
    exception_response,
)
from pyramid.i18n import make_localizer
from pyramid.interfaces import ITranslationDirectories
from pyramid.renderers import render_to_response
from pyramid.response import Response
from pyramid.view import (
    exception_view_config,
    forbidden_view_config,
    notfound_view_config,
    view_config,
    view_defaults,
)
from sqlalchemy import func
from sqlalchemy.orm import aliased, joinedload
from sqlalchemy.orm.exc import NoResultFound
from sqlalchemy.sql import exists, expression
from trove_classifiers import deprecated_classifiers, sorted_classifiers

from warehouse.accounts import REDIRECT_FIELD_NAME
from warehouse.accounts.forms import TitanPromoCodeForm
from warehouse.accounts.models import TitanPromoCode, User
from warehouse.cache.http import add_vary, cache_control
from warehouse.cache.origin import origin_cache
from warehouse.classifiers.models import Classifier
from warehouse.db import DatabaseNotAvailableError
from warehouse.errors import WarehouseDenied
from warehouse.forms import SetLocaleForm
from warehouse.i18n import LOCALE_ATTR
from warehouse.manage.views import user_projects
from warehouse.metrics import IMetricsService
from warehouse.packaging.models import File, Project, Release, release_classifiers
from warehouse.search.queries import SEARCH_FILTER_ORDER, get_es_query
from warehouse.utils.http import is_safe_url
from warehouse.utils.paginate import ElasticsearchPage, paginate_url_factory
from warehouse.utils.row_counter import RowCount

JSON_REGEX = r"^/pypi/([^\/]+)\/?([^\/]+)?/json\/?$"
json_path = re.compile(JSON_REGEX)


@view_config(context=HTTPException)
@notfound_view_config(append_slash=HTTPMovedPermanently)
@notfound_view_config(path_info=JSON_REGEX, append_slash=False)
def httpexception_view(exc, request):
    # This special case exists for the easter egg that appears on the 404
    # response page. We don't generally allow youtube embeds, but we make an
    # except for this one.
    if isinstance(exc, HTTPNotFound):
        request.find_service(name="csp").merge(
            {
                "frame-src": ["https://www.youtube-nocookie.com"],
                "script-src": ["https://www.youtube.com", "https://s.ytimg.com"],
            }
        )
    try:
        # Lightweight version of 404 page for `/simple/`
        if isinstance(exc, HTTPNotFound) and request.path.startswith("/simple/"):
            response = Response(body="404 Not Found", content_type="text/plain")
        elif isinstance(exc, HTTPNotFound) and json_path.match(request.path):
            response = Response(
                body='{"message": "Not Found"}',
                charset="utf-8",
                content_type="application/json",
            )
        else:
            response = render_to_response(
                "{}.html".format(exc.status_code), {}, request=request
            )
    except LookupError:
        # We don't have a customized template for this error, so we'll just let
        # the default happen instead.
        return exc

    # Copy over the important values from our HTTPException to our new response
    # object.
    response.status = exc.status
    response.headers.extend(
        (k, v) for k, v in exc.headers.items() if k not in response.headers
    )

    return response


@forbidden_view_config()
@exception_view_config(PredicateMismatch)
def forbidden(exc, request):
    # If the forbidden error is because the user isn't logged in, then we'll
    # redirect them to the log in page.
    if request.authenticated_userid is None:
        url = request.route_url(
            "accounts.login", _query={REDIRECT_FIELD_NAME: request.path_qs}
        )
        return HTTPSeeOther(url)

    # Check if the error has a "result" attribute and if it is a WarehouseDenied
    if hasattr(exc, "result") and isinstance(exc.result, WarehouseDenied):
        # If the forbidden error is because the user doesn't have 2FA enabled, we'll
        # redirect them to the 2FA page
        if exc.result.reason in {"owners_require_2fa", "pypi_mandates_2fa"}:
            request.session.flash(
                request._(
                    "Two-factor authentication must be enabled on your account to "
                    "perform this action."
                ),
                queue="error",
            )
            url = request.route_url(
                "manage.account.two-factor",
                _query={REDIRECT_FIELD_NAME: request.path_qs},
            )
            return HTTPSeeOther(url)

    # If we've reached here, then the user is logged in and they are genuinely
    # not allowed to access this page.
    return httpexception_view(exc, request)


@forbidden_view_config(path_info=r"^/_includes/")
@exception_view_config(PredicateMismatch, path_info=r"^/_includes/")
def forbidden_include(exc, request):
    # If the forbidden error is for a client-side-include, just return an empty
    # response instead of redirecting
    return Response(status=403)


@view_config(context=DatabaseNotAvailableError)
def service_unavailable(exc, request):
    return httpexception_view(HTTPServiceUnavailable(), request)


@view_config(
    route_name="robots.txt",
    renderer="robots.txt",
    decorator=[
        cache_control(1 * 24 * 60 * 60),  # 1 day
        origin_cache(
            1 * 24 * 60 * 60,  # 1 day
            stale_while_revalidate=6 * 60 * 60,  # 6 hours
            stale_if_error=1 * 24 * 60 * 60,  # 1 day
        ),
    ],
)
def robotstxt(request):
    request.response.content_type = "text/plain"
    return {}


@view_config(
    route_name="opensearch.xml",
    renderer="opensearch.xml",
    decorator=[
        cache_control(1 * 24 * 60 * 60),  # 1 day
        origin_cache(
            1 * 24 * 60 * 60,  # 1 day
            stale_while_revalidate=6 * 60 * 60,  # 6 hours
            stale_if_error=1 * 24 * 60 * 60,  # 1 day
        ),
    ],
)
def opensearchxml(request):
    request.response.content_type = "text/xml"
    return {}


@view_config(
    route_name="index",
    renderer="index.html",
    decorator=[
        origin_cache(
            1 * 60 * 60,  # 1 hour
            stale_while_revalidate=10 * 60,  # 10 minutes
            stale_if_error=1 * 24 * 60 * 60,  # 1 day
            keys=["all-projects", "trending"],
        )
    ],
    has_translations=True,
)
def index(request):
    project_ids = [
        r[0]
        for r in (
            request.db.query(Project.id)
            .order_by(Project.zscore.desc().nullslast(), func.random())
            .limit(5)
            .all()
        )
    ]
    release_a = aliased(
        Release,
        request.db.query(Release)
        .distinct(Release.project_id)
        .filter(Release.project_id.in_(project_ids))
        .order_by(
            Release.project_id,
            Release.is_prerelease.nullslast(),
            Release._pypi_ordering.desc(),
        )
        .subquery(),
    )
    trending_projects = (
        request.db.query(release_a)
        .options(joinedload(release_a.project))
        .order_by(func.array_idx(project_ids, release_a.project_id))
        .all()
    )

    latest_releases = (
        request.db.query(Release)
        .options(joinedload(Release.project))
        .order_by(Release.created.desc())
        .limit(5)
        .all()
    )

    counts = dict(
        request.db.query(RowCount.table_name, RowCount.count)
        .filter(
            RowCount.table_name.in_(
                [
                    Project.__tablename__,
                    Release.__tablename__,
                    File.__tablename__,
                    User.__tablename__,
                ]
            )
        )
        .all()
    )

    return {
        "latest_releases": latest_releases,
        "trending_projects": trending_projects,
        "num_projects": counts.get(Project.__tablename__, 0),
        "num_releases": counts.get(Release.__tablename__, 0),
        "num_files": counts.get(File.__tablename__, 0),
        "num_users": counts.get(User.__tablename__, 0),
    }


@view_config(
    route_name="locale",
    request_method="GET",
    request_param=SetLocaleForm.__params__,
    uses_session=True,
)
def locale(request):
    try:
        form = SetLocaleForm(locale_id=request.GET.getone("locale_id"))
    except KeyError:
        raise HTTPBadRequest("Invalid amount of locale_id parameters provided")

    redirect_to = request.referer
    if not is_safe_url(redirect_to, host=request.host):
        redirect_to = request.route_path("index")
    resp = HTTPSeeOther(redirect_to)

    if form.validate():
        # Build a localizer for the locale we're about to switch to. This will
        # happen automatically once the cookie is set, but if we want the flash
        # message indicating success to be in the new language as well, we need
        # to do it here.
        tdirs = request.registry.queryUtility(ITranslationDirectories)
        _ = make_localizer(form.locale_id.data, tdirs).translate
        request.session.flash(_("Locale updated"), queue="success")
        resp.set_cookie(LOCALE_ATTR, form.locale_id.data)

    return resp


@view_config(
    route_name="classifiers", renderer="pages/classifiers.html", has_translations=True
)
def list_classifiers(request):
    return {"classifiers": sorted_classifiers}


@view_config(
    route_name="search",
    renderer="search/results.html",
    decorator=[
        origin_cache(
            1 * 60 * 60,  # 1 hour
            stale_if_error=1 * 24 * 60 * 60,  # 1 day
            keys=["all-projects"],
        )
    ],
    has_translations=True,
)
def search(request):
    metrics = request.find_service(IMetricsService, context=None)

    querystring = request.params.get("q", "").replace("'", '"')
    order = request.params.get("o", "")
    classifiers = request.params.getall("c")
    query = get_es_query(request.es, querystring, order, classifiers)

    try:
        page_num = int(request.params.get("page", 1))
    except ValueError:
        raise HTTPBadRequest("'page' must be an integer.")

    try:
        page = ElasticsearchPage(
            query, page=page_num, url_maker=paginate_url_factory(request)
        )
    except elasticsearch.TransportError:
        metrics.increment("warehouse.views.search.error")
        raise HTTPServiceUnavailable

    if page.page_count and page_num > page.page_count:
        raise HTTPNotFound

    available_filters = collections.defaultdict(list)

    classifiers_q = (
        request.db.query(Classifier)
        .with_entities(Classifier.classifier)
        .filter(
            exists([release_classifiers.c.trove_id]).where(
                release_classifiers.c.trove_id == Classifier.id
            ),
            Classifier.classifier.notin_(deprecated_classifiers.keys()),
        )
        .order_by(
            expression.case(
                {c: i for i, c in enumerate(sorted_classifiers)},
                value=Classifier.classifier,
            )
        )
    )

    for cls in classifiers_q:
        first, *_ = cls.classifier.split(" :: ")
        available_filters[first].append(cls.classifier)

    def filter_key(item):
        try:
            return 0, SEARCH_FILTER_ORDER.index(item[0]), item[0]
        except ValueError:
            return 1, 0, item[0]

    def form_filters_tree(split_list):
        """
        Takes a list of lists, each of them containing a filter and
        one of its children.
        Returns a dictionary, each key being a filter and each value being
        the filter's children.
        """
        d = {}
        for list_ in split_list:
            current_level = d
            for part in list_:
                if part not in current_level:
                    current_level[part] = {}
                current_level = current_level[part]
        return d

    def process_available_filters():
        """
        Processes available filters and returns a list of dictionaries.
        The value of a key in the dictionary represents its children
        """
        sorted_filters = sorted(available_filters.items(), key=filter_key)
        output = []
        for f in sorted_filters:
            classifier_list = f[1]
            split_list = [i.split(" :: ") for i in classifier_list]
            tree = form_filters_tree(split_list)
            output.append(tree)
        return output

    metrics = request.find_service(IMetricsService, context=None)
    metrics.histogram("warehouse.views.search.results", page.item_count)

    return {
        "page": page,
        "term": querystring,
        "order": order,
        "available_filters": process_available_filters(),
        "applied_filters": request.params.getall("c"),
    }


@view_config(
    route_name="stats",
    renderer="pages/stats.html",
    decorator=[
        add_vary("Accept"),
        cache_control(1 * 24 * 60 * 60),  # 1 day
        origin_cache(
            1 * 24 * 60 * 60,  # 1 day
            stale_while_revalidate=1 * 24 * 60 * 60,  # 1 day
            stale_if_error=1 * 24 * 60 * 60,  # 1 day
        ),
    ],
    has_translations=True,
)
@view_config(
    route_name="stats.json",
    renderer="json",
    decorator=[
        add_vary("Accept"),
        cache_control(1 * 24 * 60 * 60),  # 1 day
        origin_cache(
            1 * 24 * 60 * 60,  # 1 day
            stale_while_revalidate=1 * 24 * 60 * 60,  # 1 day
            stale_if_error=1 * 24 * 60 * 60,  # 1 day
        ),
    ],
    accept="application/json",
)
def stats(request):
    total_size = int(request.db.query(func.sum(Project.total_size)).first()[0])
    top_100_packages = (
        request.db.query(Project)
        .with_entities(Project.name, Project.total_size)
        .order_by(Project.total_size.desc().nullslast())
        .limit(100)
        .all()
    )
    # Move top packages into a dict to make JSON more self describing
    top_packages = {
        pkg_name: {"size": int(pkg_bytes) if pkg_bytes is not None else 0}
        for pkg_name, pkg_bytes in top_100_packages
    }

    return {"total_packages_size": total_size, "top_packages": top_packages}


@view_defaults(
    route_name="security-key-giveaway",
    renderer="pages/security-key-giveaway.html",
    uses_session=True,
    has_translations=True,
    require_csrf=True,
    require_methods=False,
)
class SecurityKeyGiveaway:
    def __init__(self, request):
        self.request = request

    @property
    def form(self):
        return TitanPromoCodeForm(**self.request.POST)

    @property
    def codes_available(self):
        return (
            self.request.db.query(TitanPromoCode)
            .filter(TitanPromoCode.user_id.is_(None))
            .count()
        ) > 0

    @property
    def eligible_projects(self):
        if self.request.user:
            return set(
                project.name
                for project in user_projects(self.request)["projects_requiring_2fa"]
                if project.pypi_mandates_2fa
            )
        else:
            return set()

    @property
    def promo_code(self):
        if self.request.user:
            try:
                return (
                    self.request.db.query(TitanPromoCode).filter(
                        TitanPromoCode.user_id == self.request.user.id
                    )
                ).one()
            except NoResultFound:
                pass
        return None

    @property
    def default_response(self):
        codes_available = self.codes_available
        eligible_projects = self.eligible_projects
        promo_code = self.promo_code
        has_webauthn = self.request.user and self.request.user.has_webauthn

        eligible = (
            codes_available
            and bool(eligible_projects)
            and not has_webauthn
            and not promo_code
        )

        if not codes_available:
            reason_ineligible = "At this time there are no keys available"
        elif not eligible_projects:
            reason_ineligible = "You are not a collaborator on any critical projects"
        elif has_webauthn:
            reason_ineligible = (
                "You already have two-factor authentication enabled with a hardware "
                "security key"
            )
        elif promo_code:
            reason_ineligible = "Promo code has already been generated"
        else:
            reason_ineligible = None

        return {
            "eligible": eligible,
            "reason_ineligible": reason_ineligible,
            "form": self.form,
            "codes_available": self.codes_available,
            "eligible_projects": self.eligible_projects,
            "promo_code": self.promo_code,
            "REDIRECT_FIELD_NAME": REDIRECT_FIELD_NAME,
        }

    @view_config(request_method="GET")
    def security_key_giveaway(self):
        if not self.request.registry.settings.get(
            "warehouse.two_factor_mandate.available"
        ):
            raise HTTPNotFound

        return self.default_response

    @view_config(request_method="POST")
    def security_key_giveaway_submit(self):
        if not self.request.registry.settings.get(
            "warehouse.two_factor_mandate.available"
        ):
            raise HTTPNotFound

        default_response = self.default_response

        if not self.form.validate():
            self.request.session.flash("Form is not valid")
        elif not default_response["eligible"]:
            self.request.session.flash(default_response["reason_ineligible"])
        else:
            # The form is valid, assign a promo code to the user
            promo_code = (
                self.request.db.query(TitanPromoCode).filter(
                    TitanPromoCode.user_id.is_(None)
                )
            ).first()
            promo_code.user_id = self.request.user.id
            promo_code.distributed = datetime.datetime.now()
            # Flush so the promo code is available for the response
            self.request.db.flush()
            default_response["promo_code"] = promo_code
        return default_response


@view_config(
    route_name="includes.current-user-indicator",
    renderer="includes/current-user-indicator.html",
    uses_session=True,
    has_translations=True,
)
def current_user_indicator(request):
    return {}


@view_config(
    route_name="includes.flash-messages",
    renderer="includes/flash-messages.html",
    uses_session=True,
    has_translations=True,
)
def flash_messages(request):
    return {}


@view_config(
    route_name="includes.session-notifications",
    renderer="includes/session-notifications.html",
    uses_session=True,
    has_translations=True,
)
def session_notifications(request):
    return {}


@view_config(
    route_name="includes.sidebar-sponsor-logo",
    renderer="includes/sidebar-sponsor-logo.html",
    uses_session=False,
    has_translations=False,
    decorator=[
        cache_control(30),  # 30 seconds
    ],
)
def sidebar_sponsor_logo(request):
    return {}


@view_config(route_name="health", renderer="string")
def health(request):
    # This will ensure that we can access the database and run queries against
    # it without doing anything that will take a lock or block other queries.
    request.db.execute("SELECT 1")

    # Nothing will actually check this, but it's a little nicer to have
    # something to return besides an empty body.
    return "OK"


@view_config(route_name="force-status")
def force_status(request):
    try:
        raise exception_response(int(request.matchdict["status"]))
    except KeyError:
        raise exception_response(404) from None

warehouse/views.py
diff --git a/tests/unit/test_views.py b/tests/unit/test_views.py
index 1e1607d0ed11..583fd9c3a034 100644
--- a/tests/unit/test_views.py
+++ b/tests/unit/test_views.py
@@ -594,11 +594,16 @@ def test_eligible_projects_no_user(self, db_request):
 
     def test_eligible_projects_owners_require_2fa(self, db_request):
         db_request.user = UserFactory.create()
+
         ProjectFactory.create()
-        project = ProjectFactory.create(owners_require_2fa=True)
-        RoleFactory.create(user=db_request.user, project=project)
+        ProjectFactory.create(owners_require_2fa=True)
+        p1 = ProjectFactory.create(pypi_mandates_2fa=True)
+        p2 = ProjectFactory.create(owners_require_2fa=True, pypi_mandates_2fa=True)
 
-        assert SecurityKeyGiveaway(db_request).eligible_projects == {project.name}
+        RoleFactory.create(user=db_request.user, project=p1)
+        RoleFactory.create(user=db_request.user, project=p2)
+
+        assert SecurityKeyGiveaway(db_request).eligible_projects == {p1.name, p2.name}
 
     def test_eligible_projects_pypi_mandates_2fa(self, db_request):
         db_request.user = UserFactory.create()
diff --git a/warehouse/views.py b/warehouse/views.py
index cc3bee2f8270..8ad3e9912104 100644
--- a/warehouse/views.py
+++ b/warehouse/views.py
@@ -485,6 +485,7 @@ def eligible_projects(self):
             return set(
                 project.name
                 for project in user_projects(self.request)["projects_requiring_2fa"]
+                if project.pypi_mandates_2fa
             )
         else:
             return set()

2FA giveaway lists critical + 2FA required projects, should only list critical
<!--
    NOTE: This issue should be for problems with PyPI itself, including:
    * pypi.org
    * test.pypi.org
    * files.pythonhosted.org

    This issue should NOT be for a project installed from PyPI. If you are
    having an issue with a specific package, you should reach out to the
    maintainers of that project directly instead.

    Furthermore, this issue should NOT be for any non-PyPI properties (like
    python.org, docs.python.org, etc.)

    If your problem is related to search (a new or updated project doesn't
    appear in the PyPI search results), please wait for a couple of hours
    and check again before reporting it. The search index may take some
    time to be updated.
-->

**Describe the bug**
<!-- A clear and concise description the bug -->

The 2FA key giveaway eligibility says:

"Congrats, you may be eligible to receive a promo code! The following projects you maintain are considered critical projects:"

But not all of these 15 are critical:

<img width="903" alt="image" src="https://user-images.githubusercontent.com/1324225/186908565-479f41ab-6b23-4506-8469-f164fd49b96e.png">

**Expected behavior**
<!-- A clear and concise description of what you expected to happen -->

Only list the critical projects. 

At https://pypi.org/manage/projects/ only these 4 are listed as critical: prettytable, humanize, ujson, Pillow.

**To Reproduce**
<!-- Steps to reproduce the bug, or a link to PyPI where the bug is visible -->

1. Visit https://pypi.org/security-key-giveaway/
2. Click "Log in to determine if you are eligible" and log in


**My Platform**
<!--
    Any details about your specific platform:
    * If the problem is in the browser, what browser, version, and OS?
    * If the problem is with a command-line tool, what version of that tool?
    * If the problem is with connecting to PyPI, include some details about
      your network, including SSL/TLS implementation in use, internet service
      provider, and if there are any firewalls or proxies in use.
-->

n/a

**Additional context**
<!-- Add any other context, links, etc. about the feature here. -->

I have 18 projects at https://pypi.org/manage/projects/

* 4 are critical
* 11 I've also set as 2FA required
* looks like these 4+11 are being listed as critical on the promo page, but it should just be the 4

