fix academy/token

breathecode/authenticate/admin.py

@@ -523,7 +523,9 @@ def authenticate_google(self, obj):
         current_url = f"{request.scheme}://{request.get_host()}{request.get_full_path()}"
         current_url = str(base64.urlsafe_b64encode(current_url.encode("utf-8")), "utf-8")
 
-        return format_html(f"<a href='/v1/auth/academy/google?url={current_url}'>connect google</a>")
+        return format_html(
+            f"<a rel='noopener noreferrer' target='_blank' href='/v1/auth/academy/google?academysettings=set&url={current_url}'>connect google</a>"
+        )
 
 
 @admin.register(GoogleWebhook)

breathecode/authenticate/tests/urls/tests_academy_google.py

@@ -0,0 +1,260 @@
+"""
+Test /v1/auth/subscribe
+"""
+
+import base64
+import os
+import random
+from datetime import datetime
+from unittest.mock import MagicMock, call
+from urllib.parse import urlencode
+
+import pytest
+from capyc import pytest as capy
+from django import shortcuts
+from django.http import JsonResponse
+from django.urls.base import reverse_lazy
+from django.utils import timezone
+from rest_framework import status
+from rest_framework.test import APIClient
+
+now = timezone.now()
+
+
+@pytest.fixture(autouse=True)
+def redirect_url(monkeypatch: pytest.MonkeyPatch):
+    def redirect_url(*args, **kwargs):
+
+        if args:
+            args = args[1:]
+
+        if args:
+            try:
+                kwargs["_template"] = args[0]
+            except:
+                ...
+
+            try:
+                kwargs["context"] = args[1]
+            except:
+                ...
+
+            try:
+                if args[2]:
+                    kwargs["content_type"] = args[2]
+            except:
+                ...
+
+            try:
+                if args[3]:
+                    kwargs["status"] = args[3]
+            except:
+                ...
+
+            try:
+                if args[4]:
+                    kwargs["using"] = args[4]
+            except:
+                ...
+
+        if "context" in kwargs:
+            kwargs.update(kwargs["context"])
+            del kwargs["context"]
+
+        if "academy" in kwargs:
+            kwargs["academy"] = kwargs["academy"].id
+
+        return JsonResponse(kwargs, status=kwargs["status"])
+
+    monkeypatch.setattr(
+        shortcuts,
+        "render",
+        MagicMock(side_effect=redirect_url),
+    )
+    yield
+
+
+b = os.urandom(16)
+
+
+@pytest.fixture(autouse=True)
+def setup(monkeypatch: pytest.MonkeyPatch, db):
+
+    monkeypatch.setattr("os.urandom", lambda _: b)
+    monkeypatch.setattr("breathecode.authenticate.tasks.create_user_from_invite.delay", MagicMock())
+    monkeypatch.setattr("breathecode.authenticate.tasks.async_validate_email_invite.delay", MagicMock())
+    monkeypatch.setattr("breathecode.authenticate.tasks.verify_user_invite_email.delay", MagicMock())
+    monkeypatch.setattr("rest_framework.authtoken.models.Token.generate_key", MagicMock(return_value="1234567890"))
+
+    yield
+
+
+@pytest.fixture
+def validation_res(patch_request):
+    validation_res = {
+        "quality_score": (random.random() * 0.4) + 0.6,
+        "email_quality": (random.random() * 0.4) + 0.6,
+        "is_valid_format": {
+            "value": True,
+        },
+        "is_mx_found": {
+            "value": True,
+        },
+        "is_smtp_valid": {
+            "value": True,
+        },
+        "is_catchall_email": {
+            "value": True,
+        },
+        "is_role_email": {
+            "value": True,
+        },
+        "is_disposable_email": {
+            "value": False,
+        },
+        "is_free_email": {
+            "value": True,
+        },
+    }
+    patch_request(
+        [
+            (
+                call(
+                    "get",
+                    "https://emailvalidation.abstractapi.com/v1/?api_key=None&[email protected]",
+                    params=None,
+                    timeout=10,
+                ),
+                validation_res,
+            ),
+        ]
+    )
+    return validation_res
+
+
+def test_no_auth(database: capy.Database, client: APIClient):
+    url = reverse_lazy("authenticate:academy_google")
+    response = client.get(url)
+
+    assert response.status_code == status.HTTP_302_FOUND
+    assert response.url == f'/v1/auth/view/login?attempt=1&url={base64.b64encode(url.encode("utf-8")).decode("utf-8")}'
+    assert database.list_of("authenticate.Token") == []
+
+
+def test_no_callback_url(database: capy.Database, client: APIClient, format: capy.Format):
+    model = database.create(token=1, user=1)
+    url = reverse_lazy("authenticate:academy_google") + f"?token={model.token.key}"
+    response = client.get(url, headers={"Academy": 1})
+
+    json = response.json()
+    assert json == {
+        "status": 400,
+        "_template": "message.html",
+        "MESSAGE": "no callback URL specified",
+        "BUTTON": "Back to 4Geeks",
+        "BUTTON_TARGET": "_blank",
+        "LINK": os.getenv("APP_URL"),
+    }
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert database.list_of("authenticate.Token") == [
+        format.to_obj_repr(model.token),
+    ]
+
+
+def test_redirect_to_google(database: capy.Database, client: APIClient, format: capy.Format, utc_now: datetime):
+    model = database.create(token=1, user=1)
+    print(utc_now)
+    url = reverse_lazy("authenticate:academy_google") + f"?token={model.token.key}&url=https://potato.io"
+    response = client.get(url, headers={"Academy": 1})
+
+    assert response.status_code == status.HTTP_302_FOUND
+
+    query_params = {
+        "url": "https://potato.io",
+    }
+    query_string = urlencode(query_params)
+
+    assert response.url == f"/v1/auth/google/1234567890?{query_string}"
+    assert database.list_of("authenticate.Token") == [
+        format.to_obj_repr(model.token),
+        {
+            "id": 2,
+            "key": "1234567890",
+            "user_id": model.user.id,
+            "expires_at": None,
+            "created": model.token.created,
+            "token_type": "one_time",
+        },
+    ]
+
+
+@pytest.mark.parametrize("academy_settings", ["overwrite", "set"])
+def test_no_capability_with_academy_settings(
+    database: capy.Database, client: APIClient, format: capy.Format, utc_now: datetime, academy_settings: str
+):
+    model = database.create(token=1, user=1)
+    print(utc_now)
+    url = (
+        reverse_lazy("authenticate:academy_google")
+        + f"?token={model.token.key}&url=https://potato.io&academysettings={academy_settings}"
+    )
+    response = client.get(url, headers={"Academy": 1})
+
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    json = response.json()
+    assert json == {
+        "status": 403,
+        "_template": "message.html",
+        "LINK": None,
+        "MESSAGE": "You don't have permission to access this view",
+        "BUTTON": None,
+        "BUTTON_TARGET": "_blank",
+    }
+
+    assert database.list_of("authenticate.Token") == [
+        format.to_obj_repr(model.token),
+    ]
+
+
+@pytest.mark.parametrize("academy_settings", ["overwrite", "set"])
+def test_redirect_to_google_with_academy_settings(
+    database: capy.Database, client: APIClient, format: capy.Format, utc_now: datetime, academy_settings: str
+):
+    model = database.create(
+        token=1,
+        user=1,
+        academy=1,
+        profile_academy=1,
+        role=1,
+        capability={"slug": "crud_academy_auth_settings"},
+        city=1,
+        country=1,
+    )
+    print(utc_now)
+    url = (
+        reverse_lazy("authenticate:academy_google")
+        + f"?token={model.token.key}&url=https://potato.io&academysettings={academy_settings}"
+    )
+    response = client.get(url, headers={"Academy": 1})
+
+    assert response.status_code == status.HTTP_302_FOUND
+
+    query_params = {
+        "academysettings": academy_settings,
+        "url": "https://potato.io",
+    }
+    query_string = urlencode(query_params)
+
+    assert response.url == f"/v1/auth/google/1234567890?{query_string}"
+    assert database.list_of("authenticate.Token") == [
+        format.to_obj_repr(model.token),
+        {
+            "id": 2,
+            "key": "1234567890",
+            "user_id": model.user.id,
+            "expires_at": None,
+            "created": model.token.created,
+            "token_type": "one_time",
+        },
+    ]

breathecode/authenticate/urls.py

@@ -155,7 +155,7 @@
     # google authentication oath2.0
     path("google/callback", save_google_token, name="google_callback"),
     path("google/<str:token>", get_google_token, name="google_token"),
-    path("academy/google", render_google_connect, name="academy_google_token"),
+    path("academy/google", render_google_connect, name="academy_google"),
     path("gitpod/sync", sync_gitpod_users_view, name="sync_gitpod_users"),
     # sync with gitHUB
     path("academy/github/user", GithubUserView.as_view(), name="github_user"),

breathecode/authenticate/views.py

@@ -2253,6 +2253,19 @@ async def async_iter(iterable: list):
 
 @private_view()
 def render_google_connect(request, token):
+    academy_settings = request.GET.get("academysettings", "none")
+    query = {}
+
+    if academy_settings != "none":
+        capable = ProfileAcademy.objects.filter(
+            user=request.user.id, role__capabilities__slug="crud_academy_auth_settings"
+        )
+
+        if capable.count() == 0:
+            return render_message(request, "You don't have permission to access this view", status=403)
+
+        query["academysettings"] = academy_settings
+
     callback_url = request.GET.get("url", None)
 
     if not callback_url:
@@ -2265,11 +2278,18 @@ def render_google_connect(request, token):
             callback_url = str(base64.urlsafe_b64encode(query_params.get("url", [None])[0].encode("utf-8")), "utf-8")
 
     if callback_url is None:
-        raise ValidationException("Callback URL specified", slug="no-callback")
+        extra = {}
+        if "APP_URL" in os.environ and os.getenv("APP_URL") != "":
+            extra["btn_url"] = os.getenv("APP_URL")
+            extra["btn_label"] = "Back to 4Geeks"
+
+        return render_message(request, "no callback URL specified", **extra, status=400)
+
+    query["url"] = callback_url
 
-    token, created = Token.get_or_create(user=request.user, token_type="one_time")
+    token, _ = Token.get_or_create(user=request.user, token_type="one_time")
 
-    url = f"/v1/auth/google/{token}?url={callback_url}"
+    url = f"/v1/auth/google/{token}?{urlencode(query)}"
     return HttpResponseRedirect(redirect_to=url)
 
 

breathecode/utils/decorators/capable_of.py

@@ -1,12 +1,12 @@
 from asgiref.sync import sync_to_async
+from capyc.rest_framework.exceptions import ValidationException
 from django.contrib.auth.models import AnonymousUser
 from rest_framework.exceptions import PermissionDenied
 from rest_framework.views import APIView
 
 from breathecode.utils.exceptions import ProgrammingError
-from capyc.rest_framework.exceptions import ValidationException
 
-__all__ = ["capable_of", "acapable_of"]
+__all__ = ["capable_of", "acapable_of", "get_academy_from_capability"]
 
 
 def capable_of(capability=None):

breathecode/utils/decorators/has_permission.py

@@ -5,14 +5,14 @@
 
 from adrf.requests import AsyncRequest
 from asgiref.sync import sync_to_async
+from capyc.rest_framework.exceptions import PaymentException, ValidationException
 from django.contrib.auth.models import AnonymousUser
 from django.http import HttpRequest, HttpResponse, JsonResponse
 from django.shortcuts import render
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
 from breathecode.authenticate.models import Permission, User
-from capyc.rest_framework.exceptions import PaymentException, ValidationException
 
 from ..exceptions import ProgrammingError