Merge branch 'main' into Ajay/milestone-3-migration

Author: ajay-sentry

codecov/commands/base.py

@@ -1,7 +1,16 @@
+from django.conf import settings
 from django.contrib.auth.models import AnonymousUser
 
-from codecov.commands.exceptions import MissingService
+import services.self_hosted as self_hosted
+from codecov.commands.exceptions import (
+    MissingService,
+    Unauthenticated,
+    Unauthorized,
+    ValidationError,
+)
+from codecov_auth.helpers import current_user_part_of_org
 from codecov_auth.models import Owner, User
+from core.models import Repository
 
 
 class BaseCommand:
@@ -44,3 +53,59 @@ def __init__(self, current_owner: Owner, service: str, current_user: User = None
 
         if self.current_owner:
             self.current_user = self.current_owner.user
+
+    def ensure_is_admin(self, owner: Owner) -> None:
+        """
+        Ensures that the `current_owner` is an admin of `owner`,
+        or raise `Unauthorized` otherwise.
+        """
+
+        if not current_user_part_of_org(self.current_owner, owner):
+            raise Unauthorized()
+
+        if settings.IS_ENTERPRISE:
+            if not self_hosted.is_admin_owner(self.current_owner):
+                raise Unauthorized()
+        else:
+            if not owner.is_admin(self.current_owner):
+                raise Unauthorized()
+
+    def resolve_owner_and_repo(
+        self,
+        owner_username: str,
+        repo_name: str,
+        ensure_is_admin: bool = False,
+        only_viewable: bool = False,
+        only_active: bool = False,
+    ) -> tuple[Owner, Repository]:
+        """
+        Resolves the `Owner` and `Repository` based on the passed `owner_username`
+        and `repo_name` respectively.
+
+        If `ensure_is_admin` is set, this will also ensure that the `current_owner` is an
+        admin on the resolved `Owner`.
+        """
+        if ensure_is_admin and not self.current_user.is_authenticated:
+            raise Unauthenticated()
+
+        owner = Owner.objects.filter(
+            service=self.service, username=owner_username
+        ).first()
+
+        if not owner:
+            raise ValidationError("Owner not found")
+
+        if ensure_is_admin:
+            self.ensure_is_admin(owner)
+
+        repo_query = Repository.objects
+        if only_viewable:
+            repo_query = repo_query.viewable_repos(self.current_owner)
+        if only_active:
+            repo_query = repo_query.filter(active=True)
+
+        repo = repo_query.filter(author=owner, name=repo_name).first()
+        if not repo:
+            raise ValidationError("Repo not found")
+
+        return (owner, repo)

codecov_auth/tests/unit/views/test_base.py

@@ -1,4 +1,4 @@
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from unittest.mock import Mock, patch
 
 import pytest
@@ -8,10 +8,15 @@
 from django.http import HttpResponse
 from django.test import RequestFactory, TestCase, override_settings
 from freezegun import freeze_time
-from shared.django_apps.codecov_auth.tests.factories import OwnerFactory, UserFactory
+from shared.django_apps.codecov_auth.tests.factories import (
+    OwnerFactory,
+    SessionFactory,
+    UserFactory,
+)
 from shared.license import LicenseInformation
 
-from codecov_auth.models import Owner, OwnerProfile
+from codecov_auth.models import DjangoSession, Owner, OwnerProfile, Session
+from codecov_auth.tests.factories import DjangoSessionFactory
 from codecov_auth.views.base import LoginMixin, StateMixin
 
 
@@ -729,3 +734,118 @@ def test_login_authenticated_with_claimed_owner(self):
         # does not re-claim owner
         assert owner.user is not None
         assert owner.user != user
+
+    @patch("services.refresh.RefreshService.trigger_refresh", lambda *args: None)
+    def test_login_owner_with_expired_login_session(self):
+        user = UserFactory()
+        owner = OwnerFactory(service="github", user=user)
+
+        another_user = UserFactory()
+        another_owner = OwnerFactory(service="github", user=another_user)
+
+        now = datetime.now(timezone.utc)
+
+        # Create a session that will be deleted
+        to_be_deleted_1 = SessionFactory(
+            owner=owner,
+            type="login",
+            name="to_be_deleted",
+            lastseen="2021-01-01T00:00:00+00:00",
+            login_session=DjangoSessionFactory(expire_date=now - timedelta(days=1)),
+        )
+        to_be_deleted_1_session_key = to_be_deleted_1.login_session.session_key
+
+        # Create a session that will not be deleted because its not a login session
+        to_be_kept_1 = SessionFactory(
+            owner=owner,
+            type="api",
+            name="to_be_kept",
+            lastseen="2021-01-01T00:00:00+00:00",
+            login_session=DjangoSessionFactory(expire_date=now + timedelta(days=1)),
+        )
+
+        # Create a session that will not be deleted because it's not expired
+        to_be_kept_2 = SessionFactory(
+            owner=owner,
+            type="login",
+            name="to_be_kept",
+            lastseen="2021-01-01T00:00:00+00:00",
+            login_session=DjangoSessionFactory(expire_date=now + timedelta(days=1)),
+        )
+
+        # Create a session that will not be deleted because it's not the owner's session
+        to_be_kept_3 = SessionFactory(
+            owner=another_owner,
+            type="login",
+            name="to_be_kept",
+            lastseen="2021-01-01T00:00:00+00:00",
+            login_session=DjangoSessionFactory(expire_date=now - timedelta(seconds=1)),
+        )
+
+        assert (
+            len(DjangoSession.objects.filter(session_key=to_be_deleted_1_session_key))
+            == 1
+        )
+        assert (
+            len(
+                DjangoSession.objects.filter(
+                    session_key=to_be_kept_1.login_session.session_key
+                )
+            )
+            == 1
+        )
+        assert (
+            len(
+                DjangoSession.objects.filter(
+                    session_key=to_be_kept_2.login_session.session_key
+                )
+            )
+            == 1
+        )
+        assert (
+            len(
+                DjangoSession.objects.filter(
+                    session_key=to_be_kept_3.login_session.session_key
+                )
+            )
+            == 1
+        )
+
+        self.request.user = user
+        self.mixin_instance.login_owner(owner, self.request, HttpResponse())
+        owner.refresh_from_db()
+
+        new_login_session = Session.objects.filter(name=None)
+
+        assert len(new_login_session) == 1
+        assert len(Session.objects.filter(name="to_be_deleted").all()) == 0
+        assert len(Session.objects.filter(name="to_be_kept").all()) == 3
+
+        assert (
+            len(DjangoSession.objects.filter(session_key=to_be_deleted_1_session_key))
+            == 0
+        )
+        assert (
+            len(
+                DjangoSession.objects.filter(
+                    session_key=to_be_kept_1.login_session.session_key
+                )
+            )
+            == 1
+        )
+        assert (
+            len(
+                DjangoSession.objects.filter(
+                    session_key=to_be_kept_2.login_session.session_key
+                )
+            )
+            == 1
+        )
+        assert (
+            len(
+                DjangoSession.objects.filter(
+                    session_key=to_be_kept_3.login_session.session_key
+                )
+            )
+            == 1
+        )