[SDAN-752] Allow Basic auth for ATOM and RSS

newsroom/news_api/api_tokens/auth.py

@@ -1,4 +1,5 @@
 import ipaddress
+import base64
 from typing import Any
 from datetime import timedelta
 from quart_babel import gettext
@@ -48,22 +49,41 @@ async def support_auth_token_in_url(request: Request) -> None:
     return None
 
 
+async def support_auth_basic_auth(request: Request) -> None:
+    """Flag to indicate this endpoint supports using the basic auth"""
+    return None
+
+
 class CompanyTokenAuth(UserAuthProtocol):
     def get_token_from_request(self, request: Request) -> str | None:
         """
         Extracts the token from `Authorization` header. Code taken partly
         from eve.Auth module
         """
+        supports_basic_auth = (
+            isinstance(request.endpoint.auth, list) and support_auth_basic_auth in request.endpoint.auth
+        ) or (isinstance(request.endpoint.auth, dict) and support_auth_basic_auth in request.endpoint.auth.values())
+        supports_token_in_url = (
+            isinstance(request.endpoint.auth, list) and support_auth_token_in_url in request.endpoint.auth
+        ) or (isinstance(request.endpoint.auth, dict) and support_auth_token_in_url in request.endpoint.auth.values())
 
         auth = (request.get_header("Authorization") or "").strip()
         if len(auth):
             if auth.lower().startswith(("token", "bearer")):
                 return auth.split(" ")[1] if " " in auth else None
+
+            if supports_basic_auth and auth.lower().startswith("basic"):
+                base64_payload = auth.split(" ", 1)[1]
+
+                credentials_bytes = base64.b64decode(base64_payload)
+                credentials = credentials_bytes.decode("utf-8")
+                username, password = credentials.split(":", 1)
+
+                return username if username else password
+
+            # In case just the token was passed
             return auth
 
-        supports_token_in_url = (
-            isinstance(request.endpoint.auth, list) and support_auth_token_in_url in request.endpoint.auth
-        ) or (isinstance(request.endpoint.auth, dict) and support_auth_token_in_url in request.endpoint.auth.values())
         if supports_token_in_url:
             token = request.get_url_arg("token") or request.get_view_args("token")
             if token:

newsroom/news_api/news/assets/views.py

@@ -6,7 +6,7 @@
 
 from newsroom.assets import get_upload
 from newsroom.news_api.utils import post_api_audit
-from newsroom.news_api.api_tokens.auth import support_auth_token_in_url
+from newsroom.news_api.api_tokens.auth import support_auth_token_in_url, support_auth_basic_auth
 
 assets_endpoints = EndpointGroup("assets", __name__)
 
@@ -26,7 +26,7 @@ class RouteArguments(BaseModel):
     "assets/<path:asset_id>/<item_id>",
     title="Download Asset (with Wire ID)",
     methods=["GET"],
-    auth=[support_auth_token_in_url],
+    auth=[support_auth_token_in_url, support_auth_basic_auth],
 )
 async def download(args: RouteArguments, params: RouteParams, request: Request):
     """
@@ -42,7 +42,7 @@ async def download(args: RouteArguments, params: RouteParams, request: Request):
     "assets/<string:asset_id>",
     title="Download Asset",
     methods=["GET"],
-    auth=[support_auth_token_in_url],
+    auth=[support_auth_token_in_url, support_auth_basic_auth],
 )
 async def get_item(args: RouteArguments, params: RouteParams, request: Request):
     """

newsroom/news_api/news/atom/views.py

@@ -1,7 +1,7 @@
 from superdesk.core.web import EndpointGroup
 from superdesk.core.types import Request, BaseModel, Response
 
-from newsroom.news_api.api_tokens.auth import support_auth_token_in_url
+from newsroom.news_api.api_tokens.auth import support_auth_token_in_url, support_auth_basic_auth
 from newsroom.news_api.formatters import AtomFormatter
 
 atom_endpoints = EndpointGroup("atom", __name__)
@@ -25,7 +25,7 @@ async def get_atom_token(args: AtomArgs, params: None, request: Request) -> Resp
     "atom",
     title="ATOM Feed (Header auth)",
     methods=["GET"],
-    auth=[support_auth_token_in_url],
+    auth=[support_auth_basic_auth, support_auth_token_in_url],
 )
 async def get_atom_authed(args: None, params: AtomArgs, request: Request) -> Response:
     return await AtomFormatter().format_feed(params.token, request)

newsroom/news_api/news/item/item.py

@@ -13,7 +13,7 @@
 from newsroom.settings import get_setting
 from newsroom.news_api.utils import post_api_audit
 from newsroom.history_async import HistoryService
-
+from newsroom.news_api.api_tokens.auth import support_auth_token_in_url, support_auth_basic_auth
 
 news_item_endpoints = EndpointGroup("news/item", __name__)
 
@@ -26,7 +26,12 @@ class RouteParams(BaseModel):
     format: str = "NINJSFormatter"
 
 
-@news_item_endpoints.endpoint("news/item/<path:item_id>", title="Get News Item", methods=["GET"])
+@news_item_endpoints.endpoint(
+    "news/item/<path:item_id>",
+    title="Get News Item",
+    methods=["GET"],
+    auth=[support_auth_token_in_url, support_auth_basic_auth],
+)
 async def get_item(args: RouteArguments, params: RouteParams, request: Request) -> Response:
     app = get_current_app()
     formatter = get_formatter_by_classname(params.format)

newsroom/news_api/news/rss/views.py

@@ -1,7 +1,7 @@
 from superdesk.core.web import EndpointGroup
 from superdesk.core.types import Request, BaseModel, Response
 
-from newsroom.news_api.api_tokens.auth import support_auth_token_in_url
+from newsroom.news_api.api_tokens.auth import support_auth_token_in_url, support_auth_basic_auth
 from newsroom.news_api.formatters import RSSFormatter
 
 rss_endpoints = EndpointGroup("rss", __name__)
@@ -25,7 +25,7 @@ async def get_rss_token(args: RSSArgs, params: None, request: Request) -> Respon
     "rss",
     title="RSS Feed (Header auth)",
     methods=["GET"],
-    auth=[support_auth_token_in_url],
+    auth=[support_auth_basic_auth, support_auth_token_in_url],
 )
 async def get_rss_authed(args: None, params: RSSArgs, request: Request) -> Response:
     return await RSSFormatter().format_feed(params.token, request)

tests/news_api/test_api_assets.py

@@ -1,3 +1,4 @@
+import base64
 import os
 from bson import ObjectId
 from tests.news_api.test_api_audit import audit_check
@@ -30,6 +31,21 @@ async def test_get_asset(client, app):
     response = await client.get("api/v1/assets/{}".format(image_id), headers={"Authorization": token.get("token")})
     assert response.status_code == 200
     await audit_check(str(image_id))
+    response = await client.get(
+        "api/v1/assets/{}".format(image_id), headers={"Authorization": f"Bearer {token.get('token')}"}
+    )
+    assert response.status_code == 200
+    response = await client.get(
+        "api/v1/assets/{}".format(image_id), headers={"Authorization": f"Token {token.get('token')}"}
+    )
+    assert response.status_code == 200
+    credentials_string = f"{token.get('token')}:password"
+    credentials_bytes = credentials_string.encode("utf-8")
+    encoded_payload = base64.b64encode(credentials_bytes).decode("utf-8")
+    response = await client.get(
+        "api/v1/assets/{}".format(image_id), headers={"Authorization": f"Basic {encoded_payload}"}
+    )
+    assert response.status_code == 200
 
 
 async def test_authorization_get_asset(client, app):

tests/news_api/test_news_api.py

@@ -1,3 +1,5 @@
+import base64
+
 from bson import ObjectId
 import lxml.etree
 from newsroom.types import SectionEnum
@@ -181,3 +183,40 @@ async def test_product_search(client, app):
         f"/api/v1/news/search/?products={wire_product_id}", headers={"Authorization": token.get("token")}
     )
     assert response.status_code == 400
+
+
+async def test_get_atom_and_rss_auth(client, app):
+    company_id = ObjectId()
+    await create_entries_for(
+        "companies",
+        [
+            {
+                "_id": company_id,
+                "name": "Test Company",
+                "is_enabled": True,
+                "products": [],
+                "sections": {"news_api": True, "wire": True},
+            }
+        ],
+    )
+    await create_entries_for("news_api_tokens", [{"company": company_id, "enabled": True}])
+    token = await find_one_for("news_api_tokens", company=company_id)
+
+    response = await client.get("/api/v1/rss", headers={"Authorization": token.get("token")})
+    assert response.status_code == 200
+    response = await client.get("/api/v1/atom", headers={"Authorization": "Bearer " + token.get("token")})
+    assert response.status_code == 200
+
+    credentials_string = f"{token.get('token')}:password"
+    credentials_bytes = credentials_string.encode("utf-8")
+    encoded_payload = base64.b64encode(credentials_bytes).decode("utf-8")
+
+    response = await client.get("api/v1/rss", headers={"Authorization": f"Basic {encoded_payload}"})
+    assert response.status_code == 200
+    response = await client.get("api/v1/atom", headers={"Authorization": f"Basic {encoded_payload}"})
+
+    assert response.status_code == 200
+    response = await client.get(f"api/v1/atom/{token.get('token')}")
+    assert response.status_code == 200
+    response = await client.get(f"api/v1/rss/{token.get('token')}")
+    assert response.status_code == 200