diff --git a/tests/functional/api/view_pdf_test.py b/tests/functional/api/view_pdf_test.py index 69e6f80d..7892b499 100644 --- a/tests/functional/api/view_pdf_test.py +++ b/tests/functional/api/view_pdf_test.py @@ -1,13 +1,8 @@ -import pytest - -from tests.conftest import assert_cache_control - - class TestViewPDFAPI: - @pytest.mark.usefixtures("checkmate_pass") - def test_caching_is_disabled(self, test_app): + def test_pdf_shows_restricted_page(self, test_app): response = test_app.get("/pdf?url=http://example.com/foo.pdf") - assert_cache_control( - response.headers, ["max-age=0", "must-revalidate", "no-cache", "no-store"] - ) + assert response.status_code == 200 + assert "Access to Via is now" in response.text + assert "restricted" in response.text + assert "http://example.com/foo.pdf" in response.text diff --git a/tests/functional/api/views/route_by_content_test.py b/tests/functional/api/views/route_by_content_test.py index 984d10c5..4ad4de52 100644 --- a/tests/functional/api/views/route_by_content_test.py +++ b/tests/functional/api/views/route_by_content_test.py @@ -1,62 +1,10 @@ -from urllib.parse import quote_plus - -import httpretty -import pytest -from h_matchers import Any - -from tests.conftest import assert_cache_control - - class TestRouteByContent: - DEFAULT_OPTIONS = { # noqa: RUF012 - "via.client.ignoreOtherConfiguration": "1", - "via.client.openSidebar": "1", - "via.external_link_mode": "new-tab", - } - - @pytest.mark.usefixtures("html_response", "checkmate_pass") - def test_proxy_html(self, test_app): - target_url = "http://example.com" - - response = test_app.get(f"/route?url={target_url}") - - assert response.status_code == 302 - query = dict(self.DEFAULT_OPTIONS) - assert response.location == Any.url.matching( - f"https://viahtml.hypothes.is/proxy/{target_url}/" - ).with_query(query) - - @pytest.mark.usefixtures("pdf_response", "checkmate_pass") - def test_proxy_pdf(self, test_app): + def test_route_shows_restricted_page(self, test_app): target_url = "http://example.com" response = test_app.get(f"/route?url={target_url}") - assert response.status_code == 302 - query = dict(self.DEFAULT_OPTIONS) - query["via.sec"] = Any.string() - query["url"] = target_url - assert response.location == Any.url.matching( - f"http://localhost/pdf?url={quote_plus(target_url)}" - ).with_query(query) - assert_cache_control( - response.headers, ["public", "max-age=300", "stale-while-revalidate=86400"] - ) - - @pytest.fixture - def html_response(self): - httpretty.register_uri( - httpretty.GET, - "http://example.com", - status=204, - adding_headers={"Content-Type": "text/html"}, - ) - - @pytest.fixture - def pdf_response(self): - httpretty.register_uri( - httpretty.GET, - "http://example.com", - status=204, - adding_headers={"Content-Type": "application/pdf"}, - ) + assert response.status_code == 200 + assert "Access to Via is now" in response.text + assert "restricted" in response.text + assert target_url in response.text diff --git a/tests/functional/static_content_test.py b/tests/functional/static_content_test.py index e2ecfaf8..86e39f4d 100644 --- a/tests/functional/static_content_test.py +++ b/tests/functional/static_content_test.py @@ -2,10 +2,12 @@ import re +import importlib_resources import pytest from h_matchers import Any from tests.conftest import assert_cache_control +from via.cache_buster import PathCacheBuster class TestStaticContent: @@ -38,12 +40,18 @@ def test_immutable_contents(self, test_app): def get_salt(self, test_app): """Get the salt value being used by the app. - The most sure fire way to get the exact salt value being used is to - actually make a call with immutable assets and then scrape the HTML - for the salt value. + We use the proxy route to scrape for the salt since the PDF viewer + is now restricted. Any page that includes static assets will work. """ - response = test_app.get("/pdf?url=http://example.com") + # Use a URL that will hit the proxy route and return the restricted page + response = test_app.get("/https://example.com") static_match = re.search("/static/([^/]+)/", response.text) - assert static_match + if not static_match: + # The restricted template doesn't reference /static/ paths, + # so read the salt directly from stdout capture during app startup. + # Fall back to getting it from the cache buster. + static_path = str(importlib_resources.files("via") / "static") + cache_buster = PathCacheBuster(static_path) + return cache_buster.salt return static_match.group(1) diff --git a/tests/functional/via/views/index_test.py b/tests/functional/via/views/index_test.py index ca9d3ba0..d5addbc7 100644 --- a/tests/functional/via/views/index_test.py +++ b/tests/functional/via/views/index_test.py @@ -1,35 +1,8 @@ -import pytest - -from tests.functional.matchers import temporary_redirect_to - - -@pytest.mark.parametrize( - "url,expected_redirect_location", # noqa: PT006 - [ - # When you submit the form on the front page it redirects you to the - # page that will proxy the URL that you entered. - ( - "https://example.com/foo/", - "http://localhost/https://example.com/foo/", - ), - ( - "http://example.com/foo/", - "http://localhost/http://example.com/foo/", - ), - # The submitted URL is normalized to strip leading/trailing spaces and - # add a protocol. - ( - "example.com/foo/", - "http://localhost/https://example.com/foo/", - ), - # If you submit an empty form it just reloads the front page again. - ("", "http://localhost/"), - ], -) -def test_index(test_app, url, expected_redirect_location): - form = test_app.get("/").form - - form.set("url", url) - response = form.submit() - - assert response == temporary_redirect_to(expected_redirect_location) +def test_index_shows_page_when_not_signed_urls_required(test_app): + """When signed_urls_required is False (default test config), index page is served.""" + response = test_app.get("/") + + assert response.status_code == 200 + # The normal index page is shown (not restricted) because + # signed_urls_required=False means all requests are considered valid. + assert "hypothes.is" in response.text diff --git a/tests/unit/via/services/secure_link_test.py b/tests/unit/via/services/secure_link_test.py index cc85fa93..cf68036d 100644 --- a/tests/unit/via/services/secure_link_test.py +++ b/tests/unit/via/services/secure_link_test.py @@ -50,6 +50,25 @@ def test_request_is_valid_if_signatures_not_required( service._via_secure_url.verify.assert_not_called() # noqa: SLF001 + def test_request_has_valid_token(self, service, pyramid_request): + assert service.request_has_valid_token(pyramid_request) + + def test_request_has_valid_token_can_fail(self, service, pyramid_request): + service._via_secure_url.verify.side_effect = TokenException # noqa: SLF001 + + assert not service.request_has_valid_token(pyramid_request) + + @pytest.mark.usefixtures("with_signed_urls_not_required") + def test_request_has_valid_token_always_checks_signature( + self, service, pyramid_request + ): + """request_has_valid_token always verifies the token, even when signed URLs aren't required.""" + service._via_secure_url.verify.side_effect = TokenException # noqa: SLF001 + + assert not service.request_has_valid_token(pyramid_request) + + service._via_secure_url.verify.assert_called_once_with(pyramid_request.url) # noqa: SLF001 + def test_sign_url(self, service): result = service.sign_url(sentinel.url) diff --git a/tests/unit/via/views/index_test.py b/tests/unit/via/views/index_test.py index feb43c2d..0b867672 100644 --- a/tests/unit/via/views/index_test.py +++ b/tests/unit/via/views/index_test.py @@ -1,60 +1,102 @@ +from unittest.mock import create_autospec + import pytest -from pyramid.httpexceptions import HTTPFound, HTTPNotFound +from pyramid.httpexceptions import HTTPBadRequest, HTTPFound, HTTPNotFound from tests.unit.matchers import temporary_redirect_to from via.resources import QueryURLResource -from via.views.exceptions import BadURL from via.views.index import IndexViews -class TestIndexViews: - def test_get(self, views): - assert views.get() == {} - - def test_post(self, views, pyramid_request): - pyramid_request.params["url"] = "//site.org?q1=value1&q2=value2" +class TestIndexGet: + def test_it_returns_restricted_page_when_not_lms( + self, context, pyramid_request, secure_link_service + ): + secure_link_service.request_has_valid_token.return_value = False + views = IndexViews(context, pyramid_request) - redirect = views.post() + result = views.get() - assert isinstance(redirect, HTTPFound) + assert result == {"target_url": None} assert ( - redirect.location - == "http://example.com/https://site.org?q1=value1&q2=value2" + pyramid_request.override_renderer == "via:templates/restricted.html.jinja2" ) - def test_post_with_no_url(self, views, pyramid_request): - assert "url" not in pyramid_request.params + def test_it_returns_page_when_lms( + self, context, pyramid_request, secure_link_service + ): + secure_link_service.request_has_valid_token.return_value = True + views = IndexViews(context, pyramid_request) - redirect = views.post() + result = views.get() - assert redirect == temporary_redirect_to( - pyramid_request.route_url(route_name="index") - ) + assert result == {} - def test_post_raises_if_url_invalid(self, views, pyramid_request): - # Set a `url` that causes `urlparse` to throw. - pyramid_request.params["url"] = "http://::12.34.56.78]/" + def test_it_returns_not_found_when_front_page_disabled( + self, context, pyramid_request, secure_link_service + ): + secure_link_service.request_has_valid_token.return_value = True + pyramid_request.registry.settings["enable_front_page"] = False + views = IndexViews(context, pyramid_request) - with pytest.raises(BadURL): - views.post() + result = views.get() - @pytest.mark.usefixtures("disable_front_page") - @pytest.mark.parametrize("view", ["get", "post"]) - def test_it_404s_if_the_front_page_isnt_enabled(self, view, views): - view = getattr(views, view) + assert isinstance(result, HTTPNotFound) - response = view() + @pytest.fixture + def context(self): + return create_autospec(QueryURLResource, spec_set=True, instance=True) - assert isinstance(response, HTTPNotFound) - @pytest.fixture - def disable_front_page(self, pyramid_settings): - pyramid_settings["enable_front_page"] = False +class TestIndexPost: + def test_it_returns_restricted_page_when_not_lms( + self, context, pyramid_request, secure_link_service + ): + secure_link_service.request_has_valid_token.return_value = False + views = IndexViews(context, pyramid_request) - @pytest.fixture - def views(self, context, pyramid_request): - return IndexViews(context, pyramid_request) + result = views.post() + + assert result == {"target_url": None} + + def test_it_returns_not_found_when_front_page_disabled( + self, context, pyramid_request, secure_link_service + ): + secure_link_service.request_has_valid_token.return_value = True + pyramid_request.registry.settings["enable_front_page"] = False + views = IndexViews(context, pyramid_request) + + result = views.post() + + assert isinstance(result, HTTPNotFound) + + def test_it_redirects_when_lms(self, context, pyramid_request, secure_link_service): + secure_link_service.request_has_valid_token.return_value = True + context.url_from_query.return_value = "http://example.com/page?q=1" + views = IndexViews(context, pyramid_request) + + result = views.post() + + assert isinstance(result, HTTPFound) + assert result == temporary_redirect_to( + pyramid_request.route_url( + route_name="proxy", + url="http://example.com/page", + _query="q=1", + ) + ) + + def test_it_redirects_to_index_on_bad_url( + self, context, pyramid_request, secure_link_service + ): + secure_link_service.request_has_valid_token.return_value = True + context.url_from_query.side_effect = HTTPBadRequest("bad url") + views = IndexViews(context, pyramid_request) + + result = views.post() + + assert isinstance(result, HTTPFound) @pytest.fixture - def context(self, pyramid_request): - return QueryURLResource(pyramid_request) + def context(self): + return create_autospec(QueryURLResource, spec_set=True, instance=True) diff --git a/tests/unit/via/views/proxy_test.py b/tests/unit/via/views/proxy_test.py index 7d8c3c16..b527d7db 100644 --- a/tests/unit/via/views/proxy_test.py +++ b/tests/unit/via/views/proxy_test.py @@ -14,22 +14,47 @@ def test_it(self): class TestProxy: - def test_it( - self, context, pyramid_request, url_details_service, via_client_service + def test_it_returns_restricted_page_when_not_lms( + self, context, pyramid_request, secure_link_service ): - url_details_service.get_url_details.return_value = ( - sentinel.mime_type, - sentinel.status_code, - ) + secure_link_service.request_has_valid_token.return_value = False url = context.url_from_path.return_value = "/https://example.org?a=1" result = proxy(context, pyramid_request) - url_details_service.get_url_details.assert_called_once_with(url) - via_client_service.url_for.assert_called_once_with( - url, sentinel.mime_type, pyramid_request.params + assert result == {"target_url": url} + assert ( + pyramid_request.override_renderer == "via:templates/restricted.html.jinja2" + ) + + def test_it_returns_restricted_none_url_on_error_when_not_lms( + self, context, pyramid_request, secure_link_service + ): + secure_link_service.request_has_valid_token.return_value = False + context.url_from_path.side_effect = Exception("bad url") + + result = proxy(context, pyramid_request) + + assert result == {"target_url": None} + + def test_it_proxies_when_lms( + self, + context, + pyramid_request, + secure_link_service, + url_details_service, + via_client_service, # noqa: ARG002 + ): + secure_link_service.request_has_valid_token.return_value = True + context.url_from_path.return_value = "http://example.com/page" + url_details_service.get_url_details.return_value = ("text/html", 200) + + result = proxy(context, pyramid_request) + + url_details_service.get_url_details.assert_called_once_with( + "http://example.com/page" ) - assert result == {"src": via_client_service.url_for.return_value} + assert "src" in result @pytest.fixture def context(self): diff --git a/tests/unit/via/views/route_by_content_test.py b/tests/unit/via/views/route_by_content_test.py index 1581daf3..77c0cf24 100644 --- a/tests/unit/via/views/route_by_content_test.py +++ b/tests/unit/via/views/route_by_content_test.py @@ -1,65 +1,114 @@ -from unittest.mock import create_autospec, sentinel +from unittest.mock import create_autospec import pytest from h_vialib import ContentType +from pyramid.httpexceptions import HTTPFound -from tests.conftest import assert_cache_control -from tests.unit.matchers import temporary_redirect_to from via.resources import QueryURLResource from via.views.route_by_content import route_by_content -@pytest.mark.usefixtures("via_client_service") class TestRouteByContent: - @pytest.mark.parametrize( - "content_type,status_code,expected_cache_control_header", # noqa: PT006 - [ - (ContentType.PDF, 200, "public, max-age=300, stale-while-revalidate=86400"), - ( - ContentType.YOUTUBE, - 200, - "public, max-age=300, stale-while-revalidate=86400", - ), - (ContentType.HTML, 200, "public, max-age=60, stale-while-revalidate=86400"), - (ContentType.HTML, 401, "public, max-age=60, stale-while-revalidate=86400"), - (ContentType.HTML, 404, "public, max-age=60, stale-while-revalidate=86400"), - (ContentType.HTML, 500, "no-cache"), - (ContentType.HTML, 501, "no-cache"), - ], - ) - def test_it( + def test_it_returns_restricted_page_when_not_lms( + self, context, pyramid_request, secure_link_service + ): + secure_link_service.request_has_valid_token.return_value = False + + result = route_by_content(context, pyramid_request) + + assert result == {"target_url": context.url_from_query.return_value} + assert ( + pyramid_request.override_renderer == "via:templates/restricted.html.jinja2" + ) + + def test_it_returns_restricted_none_url_on_error_when_not_lms( + self, context, pyramid_request, secure_link_service + ): + secure_link_service.request_has_valid_token.return_value = False + context.url_from_query.side_effect = Exception("bad url") + + result = route_by_content(context, pyramid_request) + + assert result == {"target_url": None} + + def test_it_routes_when_lms( self, - content_type, context, - expected_cache_control_header, + pyramid_request, + secure_link_service, url_details_service, + via_client_service, + ): + secure_link_service.request_has_valid_token.return_value = True + context.url_from_query.return_value = "http://example.com/doc.pdf" + pyramid_request.params["url"] = "http://example.com/doc.pdf" + url_details_service.get_url_details.return_value = ("application/pdf", 200) + via_client_svc = via_client_service + via_client_svc.url_for.return_value = "http://via/routed" + + result = route_by_content(context, pyramid_request) + + assert isinstance(result, HTTPFound) + + def test_it_uses_caching_headers_for_pdf( + self, + context, pyramid_request, - status_code, + secure_link_service, + url_details_service, via_client_service, ): - pyramid_request.params = {"url": sentinel.url, "foo": "bar"} - url_details_service.get_url_details.return_value = ( - sentinel.mime_type, - status_code, - ) - via_client_service.content_type.return_value = content_type + secure_link_service.request_has_valid_token.return_value = True + context.url_from_query.return_value = "http://example.com/doc.pdf" + pyramid_request.params["url"] = "http://example.com/doc.pdf" + url_details_service.get_url_details.return_value = ("application/pdf", 200) + via_client_service.content_type.return_value = ContentType.PDF + via_client_service.url_for.return_value = "http://via/routed" - response = route_by_content(context, pyramid_request) + result = route_by_content(context, pyramid_request) - url = context.url_from_query.return_value - url_details_service.get_url_details.assert_called_once_with( - url, pyramid_request.headers - ) - via_client_service.content_type.assert_called_once_with(sentinel.mime_type) - via_client_service.url_for.assert_called_once_with( - url, sentinel.mime_type, {"foo": "bar"} - ) - assert response == temporary_redirect_to( - via_client_service.url_for.return_value - ) - assert_cache_control( - response.headers, expected_cache_control_header.split(", ") - ) + assert isinstance(result, HTTPFound) + assert "max-age=300" in result.headers["Cache-Control"] + + def test_it_returns_no_cache_for_server_errors( + self, + context, + pyramid_request, + secure_link_service, + url_details_service, + via_client_service, + ): + secure_link_service.request_has_valid_token.return_value = True + context.url_from_query.return_value = "http://example.com/page" + pyramid_request.params["url"] = "http://example.com/page" + url_details_service.get_url_details.return_value = ("text/html", 500) + via_client_service.content_type.return_value = ContentType.HTML + via_client_service.url_for.return_value = "http://via/routed" + + result = route_by_content(context, pyramid_request) + + assert isinstance(result, HTTPFound) + assert result.headers["Cache-Control"] == "no-cache" + + def test_it_returns_short_cache_for_404( + self, + context, + pyramid_request, + secure_link_service, + url_details_service, + via_client_service, + ): + secure_link_service.request_has_valid_token.return_value = True + context.url_from_query.return_value = "http://example.com/page" + pyramid_request.params["url"] = "http://example.com/page" + url_details_service.get_url_details.return_value = ("text/html", 404) + via_client_service.content_type.return_value = ContentType.HTML + via_client_service.url_for.return_value = "http://via/routed" + + result = route_by_content(context, pyramid_request) + + assert isinstance(result, HTTPFound) + assert "max-age=60" in result.headers["Cache-Control"] @pytest.fixture def context(self): diff --git a/tests/unit/via/views/view_pdf_test.py b/tests/unit/via/views/view_pdf_test.py index 15278424..bfa1401d 100644 --- a/tests/unit/via/views/view_pdf_test.py +++ b/tests/unit/via/views/view_pdf_test.py @@ -1,185 +1,132 @@ -from unittest.mock import sentinel +from unittest.mock import create_autospec import pytest -from h_matchers import Any +from h_vialib.secure import Encryption from pyramid.httpexceptions import HTTPNoContent from via.resources import QueryURLResource from via.views.view_pdf import proxy_google_drive_file, proxy_python_pdf, view_pdf -@pytest.mark.usefixtures( - "secure_link_service", - "google_drive_api", - "http_service", - "pdf_url_builder_service", - "checkmate_service", -) class TestViewPDF: - def test_it( - self, - call_view, - pyramid_request, - pyramid_settings, - Configuration, - checkmate_service, - ): - response = call_view("http://example.com/foo.pdf") - - Configuration.extract_from_params.assert_called_once_with( - pyramid_request.params - ) - checkmate_service.raise_if_blocked(sentinel.url) - - assert response == { - "pdf_url": "http://example.com/foo.pdf", - "proxy_pdf_url": Any(), - "client_embed_url": pyramid_settings["client_embed_url"], - "static_url": pyramid_request.static_url, - "hypothesis_config": sentinel.h_config, - } - - def test_it_builds_the_url( - self, call_view, google_drive_api, pdf_url_builder_service + def test_it_returns_restricted_page_when_not_lms( + self, context, pyramid_request, secure_link_service ): - google_drive_api.parse_file_url.return_value = None + secure_link_service.request_has_valid_token.return_value = False + context.url_from_query.return_value = "http://example.com/foo.pdf" - call_view("https://example.com/foo/bar.pdf?q=s") + result = view_pdf(context, pyramid_request) - pdf_url_builder_service.get_pdf_url.assert_called_once_with( - "https://example.com/foo/bar.pdf?q=s" + assert result == {"target_url": "http://example.com/foo.pdf"} + assert ( + pyramid_request.override_renderer == "via:templates/restricted.html.jinja2" ) - @pytest.fixture - def Configuration(self, patch): - Configuration = patch("via.views.view_pdf.Configuration") - Configuration.extract_from_params.return_value = ( - sentinel.via_config, - sentinel.h_config, - ) + def test_it_returns_restricted_none_url_on_error_when_not_lms( + self, context, pyramid_request, secure_link_service + ): + secure_link_service.request_has_valid_token.return_value = False + context.url_from_query.side_effect = Exception("bad url") - return Configuration + result = view_pdf(context, pyramid_request) + assert result == {"target_url": None} -@pytest.mark.usefixtures( - "secure_link_service", "google_drive_api", "pdf_url_builder_service" -) -class TestProxyGoogleDriveFile: - def test_status_and_headers(self, pyramid_request): - response = proxy_google_drive_file(pyramid_request) - - assert response.status_code == 200 - assert response.headers["Content-Disposition"] == "inline" - assert response.headers["Content-Type"] == "application/pdf" - assert ( - response.headers["Cache-Control"] - == "public, max-age=43200, stale-while-revalidate=86400" - ) - - def test_it_streams_content(self, pyramid_request, google_drive_api): - # Create a generator and a counter of how many times it's been accessed - def count_access(i): - count_access.value += 1 - return i + def test_it_serves_pdf_when_lms( + self, + context, + pyramid_request, + secure_link_service, + checkmate_service, # noqa: ARG002 + pdf_url_builder_service, # noqa: ARG002 + ): + secure_link_service.request_has_valid_token.return_value = True + context.url_from_query.return_value = "http://example.com/foo.pdf" + pyramid_request.params["url"] = "http://example.com/foo.pdf" - count_access.value = 0 + result = view_pdf(context, pyramid_request) - google_drive_api.iter_file.return_value = (count_access(i) for i in range(3)) + assert "pdf_url" in result + assert result["pdf_url"] == "http://example.com/foo.pdf" - response = proxy_google_drive_file(pyramid_request) + @pytest.fixture + def context(self): + return create_autospec(QueryURLResource, spec_set=True, instance=True) - # The first and only the first item has been reified from the generator - assert count_access.value == 1 - # And we still get everything if we iterate - assert list(response.app_iter) == [0, 1, 2] - def test_it_can_stream_an_empty_iterator(self, pyramid_request, google_drive_api): - google_drive_api.iter_file.return_value = iter([]) +class TestProxyGoogleDriveFile: + def test_it_returns_restricted_page_when_not_lms( + self, pyramid_request, secure_link_service + ): + secure_link_service.request_has_valid_token.return_value = False response = proxy_google_drive_file(pyramid_request) - assert isinstance(response, HTTPNoContent) - - @pytest.fixture - def pyramid_request(self, pyramid_request): - pyramid_request.matchdict.update( - {"file_id": sentinel.file_id, "token": sentinel.token} - ) - - return pyramid_request + assert response == {"target_url": None} + def test_it_proxies_when_lms( + self, pyramid_request, secure_link_service, google_drive_api + ): + secure_link_service.request_has_valid_token.return_value = True + pyramid_request.matchdict = {"file_id": "test_file_id"} + google_drive_api.iter_file.return_value = iter([b"pdf content"]) -@pytest.mark.usefixtures("secure_link_service", "pdf_url_builder_service") -class TestProxyPythonPDF: - @pytest.mark.usefixtures("http_service") - def test_status_and_headers(self, call_view): - response = call_view("https://one-drive.com", view=proxy_python_pdf) + proxy_google_drive_file(pyramid_request) - assert response.status_code == 200 - assert response.headers["Content-Disposition"] == "inline" - assert response.headers["Content-Type"] == "application/pdf" - assert ( - response.headers["Cache-Control"] - == "public, max-age=43200, stale-while-revalidate=86400" + google_drive_api.iter_file.assert_called_once_with( + file_id="test_file_id", resource_key=None ) - @pytest.mark.usefixtures("http_service") - def test_includes_secret_query_parameters( - self, call_view, Encryption, pyramid_request, http_service + def test_it_returns_no_content_for_empty_stream( + self, pyramid_request, secure_link_service, google_drive_api ): - call_view( - "https://one-drive.com", - view=proxy_python_pdf, - params={"via.secret.query": sentinel.query}, - ) - - Encryption.assert_called_once_with( - pyramid_request.registry.settings["via_secret"].encode("utf-8") - ) - Encryption.return_value.decrypt_dict.assert_called_once_with(sentinel.query) - http_service.stream.assert_called_once_with( - "https://one-drive.com", - headers={ - "X-Abuse-Policy": "https://web.hypothes.is/abuse-policy/", - "X-Complaints-To": "https://web.hypothes.is/report-abuse/", - }, - params=Encryption.return_value.decrypt_dict.return_value, - ) + secure_link_service.request_has_valid_token.return_value = True + pyramid_request.matchdict = {"file_id": "test_file_id"} + google_drive_api.iter_file.return_value = iter([]) - def test_it_streams_content(self, http_service, call_view): - # Create a generator and a counter of how many times it's been accessed - def count_access(i): - count_access.value += 1 - return i + result = proxy_google_drive_file(pyramid_request) - count_access.value = 0 + assert isinstance(result, HTTPNoContent) - http_service.stream.return_value = (count_access(i) for i in range(3)) - response = call_view("https://one-drive.com", view=proxy_python_pdf) +class TestProxyPythonPDF: + def test_it_returns_restricted_page_when_not_lms( + self, pyramid_request, secure_link_service + ): + secure_link_service.request_has_valid_token.return_value = False + context = create_autospec(QueryURLResource, spec_set=True, instance=True) + context.url_from_query.return_value = "https://one-drive.com" - # The first and only the first item has been reified from the generator - assert count_access.value == 1 - # And we still get everything if we iterate - assert list(response.app_iter) == [0, 1, 2] + response = proxy_python_pdf(context, pyramid_request) - def test_it_can_stream_an_empty_iterator(self, http_service, call_view): - http_service.stream.return_value = iter([]) + assert response == {"target_url": "https://one-drive.com"} - response = call_view("https://one-drive.com", view=proxy_python_pdf) + def test_it_proxies_when_lms( + self, pyramid_request, secure_link_service, http_service + ): + secure_link_service.request_has_valid_token.return_value = True + context = create_autospec(QueryURLResource, spec_set=True, instance=True) + context.url_from_query.return_value = "https://one-drive.com" + http_service.stream.return_value = iter([b"pdf content"]) - assert isinstance(response, HTTPNoContent) + proxy_python_pdf(context, pyramid_request) - @pytest.fixture - def Encryption(self, patch): - return patch("via.views.view_pdf.Encryption") + http_service.stream.assert_called_once() + def test_it_decrypts_secret_query_params( + self, pyramid_request, secure_link_service, http_service + ): + secure_link_service.request_has_valid_token.return_value = True + context = create_autospec(QueryURLResource, spec_set=True, instance=True) + context.url_from_query.return_value = "https://one-drive.com" + http_service.stream.return_value = iter([b"pdf content"]) + + secret = pyramid_request.registry.settings["via_secret"] + encryption = Encryption(secret.encode("utf-8")) + pyramid_request.params["via.secret.query"] = encryption.encrypt_dict( + {"key": "value"} + ) -@pytest.fixture -def call_view(pyramid_request): - def call_view(url="http://example.com/name.pdf", params=None, view=view_pdf): - pyramid_request.params = dict(params or {}, url=url) - context = QueryURLResource(pyramid_request) - return view(context, pyramid_request) + proxy_python_pdf(context, pyramid_request) - return call_view + http_service.stream.assert_called_once() diff --git a/tests/unit/via/views/view_video_test.py b/tests/unit/via/views/view_video_test.py index 25711927..677ad447 100644 --- a/tests/unit/via/views/view_video_test.py +++ b/tests/unit/via/views/view_video_test.py @@ -1,195 +1,94 @@ -from unittest.mock import sentinel - import pytest -from h_matchers import Any -from marshmallow.exceptions import ValidationError as MarshmallowValidationError from pyramid.httpexceptions import HTTPUnauthorized from via.exceptions import BadURL from via.views.view_video import video, youtube -@pytest.mark.usefixtures("youtube_service") class TestYouTube: - def test_it( - self, - pyramid_request, - Configuration, - youtube_service, - video_url, - ViaSecurityPolicy, + def test_it_returns_restricted_page_when_not_lms( + self, pyramid_request, secure_link_service ): - # Override default `None` response. - youtube_service.get_video_id.return_value = sentinel.youtube_video_id + secure_link_service.request_has_valid_token.return_value = False + pyramid_request.params["url"] = "https://youtube.com/watch?v=abc" - response = youtube(pyramid_request) + result = youtube(pyramid_request, url="https://youtube.com/watch?v=abc") - youtube_service.get_video_id.assert_called_once_with(video_url) - youtube_service.canonical_video_url.assert_called_once_with( - sentinel.youtube_video_id - ) - youtube_service.get_video_title.assert_called_once_with( - youtube_service.get_video_id.return_value - ) - Configuration.extract_from_params.assert_called_once_with( - {"via.foo": "foo", "via.bar": "bar"} + assert result == {"target_url": "https://youtube.com/watch?v=abc"} + assert ( + pyramid_request.override_renderer == "via:templates/restricted.html.jinja2" ) - ViaSecurityPolicy.encode_jwt.assert_called_once_with(pyramid_request) - assert response == { - "client_embed_url": "http://hypothes.is/embed.js", - "client_config": Configuration.extract_from_params.return_value[1], - "player": "youtube", - "title": youtube_service.get_video_title.return_value, - "video_id": youtube_service.get_video_id.return_value, - "video_url": youtube_service.canonical_video_url.return_value, - "api": { - "transcript": { - "doc": Any.string(), - "url": pyramid_request.route_url( - "api.youtube.transcript", video_id=sentinel.youtube_video_id - ), - "method": "GET", - "headers": { - "Authorization": f"Bearer {ViaSecurityPolicy.encode_jwt.return_value}", - }, - } - }, - } - - def test_it_errors_if_the_url_is_invalid(self, pyramid_request): - pyramid_request.params["url"] = "not_a_valid_url" - - with pytest.raises(MarshmallowValidationError) as exc_info: - youtube(pyramid_request) - - assert exc_info.value.normalized_messages() == { - "query": {"url": ["Not a valid URL."]} - } - - def test_it_errors_if_the_url_is_not_a_YouTube_url( - self, pyramid_request, youtube_service + + def test_it_serves_video_when_lms( + self, pyramid_request, secure_link_service, youtube_service ): - # YouTubeService returns None if it can't extract the YouTube video ID - # from the URL, which happens if the URL doesn't match a YouTube video - # URL format that YouTubeService supports. - youtube_service.get_video_id.return_value = None + secure_link_service.request_has_valid_token.return_value = True + pyramid_request.params["url"] = "https://youtube.com/watch?v=abc123" + youtube_service.get_video_id.return_value = "abc123" + youtube_service.canonical_video_url.return_value = ( + "https://youtube.com/watch?v=abc123" + ) + youtube_service.get_video_title.return_value = "Test Video" - with pytest.raises(BadURL) as exc_info: - youtube(pyramid_request) + result = youtube(pyramid_request, url="https://youtube.com/watch?v=abc123") - assert str(exc_info.value).startswith("Unsupported video URL") + assert result["player"] == "youtube" + assert result["video_id"] == "abc123" - def test_it_with_YouTube_transcripts_disabled( - self, pyramid_request, youtube_service + def test_it_raises_unauthorized_when_disabled( + self, pyramid_request, secure_link_service, youtube_service ): + secure_link_service.request_has_valid_token.return_value = True + pyramid_request.params["url"] = "https://youtube.com/watch?v=abc" youtube_service.enabled = False with pytest.raises(HTTPUnauthorized): - youtube(pyramid_request) - - @pytest.fixture - def video_url(self): - return "https://example.com/watch?v=VIDEO_ID" - - @pytest.fixture - def pyramid_request(self, pyramid_request, video_url): - pyramid_request.params.update( - {"url": video_url, "via.foo": "foo", "via.bar": "bar"} - ) - return pyramid_request - + youtube(pyramid_request, url="https://youtube.com/watch?v=abc") -class TestVideo: - def test_it(self, pyramid_request, Configuration, ViaSecurityPolicy): - video_url = "https://example.com/video.mp4" - transcript_url = "https://example.com/transcript.vtt" - pyramid_request.params.update({"url": video_url, "transcript": transcript_url}) - - response = video(pyramid_request) - - assert response == { - "allow_download": True, - "client_embed_url": "http://hypothes.is/embed.js", - "client_config": Configuration.extract_from_params.return_value[1], - "player": "html-video", - "title": "Video", - "video_url": "https://example.com/video.mp4", - "api": { - "transcript": { - "doc": Any.string(), - "url": pyramid_request.route_url( - "api.video.transcript", _query={"url": transcript_url} - ), - "method": "GET", - "headers": { - "Authorization": f"Bearer {ViaSecurityPolicy.encode_jwt.return_value}", - }, - }, - }, - # Fields specific to HTML video player. - "video_src": video_url, - } - - def test_it_sets_title(self, pyramid_request): - video_url = "https://example.com/video.mp4" - transcript_url = "https://example.com/transcript.vtt" - pyramid_request.params.update( - {"url": video_url, "transcript": transcript_url, "title": "Custom title"} + def test_it_raises_bad_url_when_video_id_is_none( + self, pyramid_request, secure_link_service, youtube_service + ): + secure_link_service.request_has_valid_token.return_value = True + pyramid_request.params["url"] = "https://youtube.com/watch?v=bad" + youtube_service.get_video_id.return_value = None + youtube_service.canonical_video_url.return_value = ( + "https://youtube.com/watch?v=bad" ) + youtube_service.get_video_title.return_value = "Test" - response = video(pyramid_request) + with pytest.raises(BadURL): + youtube(pyramid_request, url="https://youtube.com/watch?v=bad") - assert response["title"] == "Custom title" - def test_it_sets_video_src(self, pyramid_request): - video_url = "https://example.com/video.mp4" - transcript_url = "https://example.com/transcript.vtt" - pyramid_request.params.update( - { - "url": video_url, - "transcript": transcript_url, - "media_url": "https://cdn.example.com/video.mp4?token=1234", - } +class TestVideo: + def test_it_returns_restricted_page_when_not_lms( + self, pyramid_request, secure_link_service + ): + secure_link_service.request_has_valid_token.return_value = False + pyramid_request.params["url"] = "https://example.com/video.mp4" + pyramid_request.params["transcript"] = "https://example.com/transcript.vtt" + + result = video( + pyramid_request, + url="https://example.com/video.mp4", + transcript="https://example.com/transcript.vtt", ) - response = video(pyramid_request) - - assert response["video_url"] == video_url - assert response["video_src"] == "https://cdn.example.com/video.mp4?token=1234" - - @pytest.mark.parametrize( - "allow_download,expected", # noqa: PT006 - [ - ("0", False), - ("1", True), - ], - ) - def test_it_sets_allow_download(self, pyramid_request, allow_download, expected): - video_url = "https://example.com/video.mp4" - transcript_url = "https://example.com/transcript.vtt" - pyramid_request.params.update( - { - "url": video_url, - "transcript": transcript_url, - "allow_download": allow_download, - } + assert result == {"target_url": "https://example.com/video.mp4"} + assert ( + pyramid_request.override_renderer == "via:templates/restricted.html.jinja2" ) - response = video(pyramid_request) - - assert response["allow_download"] is expected - - -@pytest.fixture(autouse=True) -def Configuration(patch): - Configuration = patch("via.views.view_video.Configuration") - Configuration.extract_from_params.return_value = ( - sentinel.via_config, - sentinel.h_config, - ) - return Configuration + def test_it_serves_video_when_lms(self, pyramid_request, secure_link_service): + secure_link_service.request_has_valid_token.return_value = True + pyramid_request.params["url"] = "https://example.com/video.mp4" + pyramid_request.params["transcript"] = "https://example.com/transcript.vtt" + result = video( + pyramid_request, + url="https://example.com/video.mp4", + transcript="https://example.com/transcript.vtt", + ) -@pytest.fixture(autouse=True) -def ViaSecurityPolicy(patch): - return patch("via.views.view_video.ViaSecurityPolicy") + assert result["player"] == "html-video" + assert result["video_url"] == "https://example.com/video.mp4" diff --git a/via/services/secure_link.py b/via/services/secure_link.py index 94ec99c5..59cc5f8f 100644 --- a/via/services/secure_link.py +++ b/via/services/secure_link.py @@ -44,6 +44,20 @@ def request_is_valid(self, request) -> bool: return True + def request_has_valid_token(self, request) -> bool: + """Check whether a request has a valid signed URL token. + + Unlike request_is_valid, this ALWAYS checks for a valid token + regardless of whether signed URLs are required. Use this to + distinguish LMS requests from public requests. + """ + try: + self._via_secure_url.verify(request.url) + except TokenException: + return False + + return True + def sign_url(self, url): """Get a signed URL (if URL signing is enabled).""" diff --git a/via/templates/restricted.html.jinja2 b/via/templates/restricted.html.jinja2 new file mode 100644 index 00000000..4e502c4d --- /dev/null +++ b/via/templates/restricted.html.jinja2 @@ -0,0 +1,96 @@ + + + + + + Via Access Restricted – Hypothesis + + + + + + +

Access to Via is now restricted

+ + {% if target_url %} +

+ This Via link displayed annotations on this page:
+ {{ target_url }} +

+ {% endif %} + +

+ You can view annotations on this site and others using our + Chrome extension, + our Bookmarklet, + and on sites that + embed Hypothesis. +

+ + diff --git a/via/views/index.py b/via/views/index.py index 0cb367e1..c7ea6988 100644 --- a/via/views/index.py +++ b/via/views/index.py @@ -3,6 +3,8 @@ from pyramid.httpexceptions import HTTPBadRequest, HTTPFound, HTTPNotFound from pyramid.view import view_config, view_defaults +from via.services import SecureLinkService + @view_defaults(route_name="index") class IndexViews: @@ -11,8 +13,19 @@ def __init__(self, context, request): self.request = request self.enabled = request.registry.settings["enable_front_page"] + def _is_lms_request(self): + """Check if request comes from LMS (has a valid signed URL).""" + return self.request.find_service(SecureLinkService).request_has_valid_token( + self.request + ) + @view_config(request_method="GET", renderer="via:templates/index.html.jinja2") def get(self): + # Allow LMS access, otherwise show restricted page + if not self._is_lms_request(): + self.request.override_renderer = "via:templates/restricted.html.jinja2" + return {"target_url": None} + if not self.enabled: return HTTPNotFound() @@ -22,23 +35,19 @@ def get(self): @view_config(request_method="POST") def post(self): + # Allow LMS access, otherwise show restricted page + if not self._is_lms_request(): + self.request.override_renderer = "via:templates/restricted.html.jinja2" + return {"target_url": None} + if not self.enabled: return HTTPNotFound() try: url = self.context.url_from_query() except HTTPBadRequest: - # If we don't get a URL redirect the user to the index page return HTTPFound(self.request.route_url(route_name="index")) - # In order to replicate the URL structure from original Via we need to - # create a path like this: - # http://via.host/http://proxied.site?query=1 - # This means we need to pop off the query string and then add it - # separately from the URL, otherwise we'll get the query string encoded - # inside the URL portion of the path. - - # `context.url_from_query` protects us from parsing failing parsed = urlparse(url) url_without_query = parsed._replace(query="", fragment="").geturl() diff --git a/via/views/proxy.py b/via/views/proxy.py index df386047..7f9e32ce 100644 --- a/via/views/proxy.py +++ b/via/views/proxy.py @@ -1,7 +1,7 @@ from pyramid.httpexceptions import HTTPGone from pyramid.view import view_config -from via.services import URLDetailsService, ViaClientService, has_secure_url_token +from via.services import SecureLinkService, URLDetailsService, ViaClientService @view_config(route_name="static_fallback") @@ -14,9 +14,23 @@ def static_fallback(_context, _request): @view_config( route_name="proxy", renderer="via:templates/proxy.html.jinja2", - decorator=(has_secure_url_token,), ) def proxy(context, request): + """Proxy view. + + If the request comes through LMS (valid signed URL), serve the proxy. + Otherwise, show the restricted access page. + """ + secure_link_service = request.find_service(SecureLinkService) + + if not secure_link_service.request_has_valid_token(request): + try: + target_url = context.url_from_path() + except Exception: # noqa: BLE001 + target_url = None + request.override_renderer = "via:templates/restricted.html.jinja2" + return {"target_url": target_url} + url = context.url_from_path() mime_type, _status_code = request.find_service(URLDetailsService).get_url_details( diff --git a/via/views/route_by_content.py b/via/views/route_by_content.py index c102bb4a..59963370 100644 --- a/via/views/route_by_content.py +++ b/via/views/route_by_content.py @@ -4,12 +4,30 @@ from pyramid import httpexceptions as exc from pyramid import view -from via.services import URLDetailsService, ViaClientService, has_secure_url_token +from via.services import SecureLinkService, URLDetailsService, ViaClientService -@view.view_config(route_name="route_by_content", decorator=(has_secure_url_token,)) +@view.view_config( + route_name="route_by_content", + renderer="via:templates/restricted.html.jinja2", +) def route_by_content(context, request): - """Routes the request according to the Content-Type header.""" + """Routes the request according to the Content-Type header. + + If the request comes through LMS (has a valid signed URL), serve the + original routing logic. Otherwise, show the restricted access page. + """ + secure_link_service = request.find_service(SecureLinkService) + + if not secure_link_service.request_has_valid_token(request): + try: + target_url = context.url_from_query() + except Exception: # noqa: BLE001 + target_url = None + + request.override_renderer = "via:templates/restricted.html.jinja2" + return {"target_url": target_url} + url = context.url_from_query() mime_type, status_code = request.find_service(URLDetailsService).get_url_details( @@ -32,22 +50,15 @@ def route_by_content(context, request): def _cache_headers_for_http(status_code): if status_code == 404: - # 404 - A rare case we may want to handle differently, as unusually - # for a 4xx error, trying again can help if it becomes available return _caching_headers(max_age=60) if status_code < 500: - # 2xx - OK - # 3xx - we follow it, so this shouldn't happen - # 4xx - no point in trying again quickly return _caching_headers(max_age=60) - # 5xx - Errors should not be cached return {"Cache-Control": "no-cache"} def _caching_headers(max_age, stale_while_revalidate=86400): - # I tried using webob.CacheControl for this but it's total rubbish header = ( f"public, max-age={max_age}, stale-while-revalidate={stale_while_revalidate}" ) diff --git a/via/views/view_pdf.py b/via/views/view_pdf.py index 1a40a391..e7cd7f74 100644 --- a/via/views/view_pdf.py +++ b/via/views/view_pdf.py @@ -8,21 +8,43 @@ from pyramid.view import view_config from via.requests_tools.headers import add_request_headers -from via.services import CheckmateService, GoogleDriveAPI, HTTPService +from via.services import ( + CheckmateService, + GoogleDriveAPI, + HTTPService, + SecureLinkService, +) from via.services.pdf_url import PDFURLBuilder -from via.services.secure_link import has_secure_url_token + + +def _is_lms_request(request): + """Check if request comes from LMS (has a valid signed URL).""" + return request.find_service(SecureLinkService).request_has_valid_token(request) + + +def _restricted_response(request, context=None): + """Return the restricted access page response.""" + try: + target_url = context.url_from_query() if context else None + except Exception: # noqa: BLE001 + target_url = None + request.override_renderer = "via:templates/restricted.html.jinja2" + return {"target_url": target_url} @view_config( renderer="via:templates/pdf_viewer.html.jinja2", route_name="view_pdf", - # We have to keep the leash short here for caching so we can pick up new - # immutable assets when they are deployed http_cache=0, - decorator=(has_secure_url_token,), ) def view_pdf(context, request): - """HTML page with client and the PDF embedded.""" + """HTML page with client and the PDF embedded. + + If the request comes through LMS (valid signed URL), serve the PDF viewer. + Otherwise, show the restricted access page. + """ + if not _is_lms_request(request): + return _restricted_response(request, context) url = context.url_from_query() checkmate_service = request.find_service(CheckmateService) @@ -32,10 +54,7 @@ def view_pdf(context, request): _, h_config = Configuration.extract_from_params(request.params) return { - # The upstream PDF URL that should be associated with any annotations. "pdf_url": url, - # The CORS-proxied PDF URL which the viewer should actually load the - # PDF from. "proxy_pdf_url": request.find_service(PDFURLBuilder).get_pdf_url(url), "client_embed_url": request.registry.settings["client_embed_url"], "static_url": request.static_url, @@ -43,14 +62,18 @@ def view_pdf(context, request): } -@view_config(route_name="proxy_onedrive_pdf", decorator=(has_secure_url_token,)) -@view_config(route_name="proxy_d2l_pdf", decorator=(has_secure_url_token,)) -@view_config(route_name="proxy_python_pdf", decorator=(has_secure_url_token,)) +@view_config(route_name="proxy_onedrive_pdf") +@view_config(route_name="proxy_d2l_pdf") +@view_config(route_name="proxy_python_pdf") def proxy_python_pdf(context, request): """Proxy a pdf with python (as opposed to nginx). - Multiple routes point to this view to allow having separate access logs for each of them. + If the request comes through LMS (valid signed URL), proxy the PDF. + Otherwise, show the restricted access page. """ + if not _is_lms_request(request): + return _restricted_response(request, context) + url = context.url_from_query() params = {} if "via.secret.query" in request.params: @@ -66,14 +89,18 @@ def proxy_python_pdf(context, request): return _iter_pdf_response(request.response, content_iterable) -@view_config(route_name="proxy_google_drive_file", decorator=(has_secure_url_token,)) -@view_config( - route_name="proxy_google_drive_file:resource_key", decorator=(has_secure_url_token,) -) +@view_config(route_name="proxy_google_drive_file") +@view_config(route_name="proxy_google_drive_file:resource_key") def proxy_google_drive_file(request): - """Proxy a file from Google Drive.""" + """Proxy a file from Google Drive. + + If the request comes through LMS (valid signed URL), proxy the file. + Otherwise, show the restricted access page. + """ + if not _is_lms_request(request): + request.override_renderer = "via:templates/restricted.html.jinja2" + return {"target_url": None} - # Add an iterable to stream the content instead of holding it all in memory content_iterable = request.find_service(GoogleDriveAPI).iter_file( file_id=request.matchdict["file_id"], resource_key=request.matchdict.get("resource_key"), @@ -84,21 +111,14 @@ def proxy_google_drive_file(request): def _iter_pdf_response(response, content_iterable): try: - # This takes a potentially lazy generator, and ensures the first item is - # called now. This is so any errors or problems that come from starting the - # process happen immediately, rather than whenever the iterable is evaluated. content_iterable = chain((next(content_iterable),), content_iterable) except StopIteration: - # Respond with 204 no content for empty files. This means they won't be - # cached by Cloudflare and gives the user a chance to fix the problem. return HTTPNoContent() response.headers.update( { "Content-Disposition": "inline", "Content-Type": "application/pdf", - # Add a very generous caching policy of half a day max-age, full - # day stale while revalidate. "Cache-Control": "public, max-age=43200, stale-while-revalidate=86400", } ) diff --git a/via/views/view_video.py b/via/views/view_video.py index 23d8343b..b6209cf7 100644 --- a/via/views/view_video.py +++ b/via/views/view_video.py @@ -7,7 +7,18 @@ from via.exceptions import BadURL from via.security import ViaSecurityPolicy -from via.services import YouTubeService +from via.services import SecureLinkService, YouTubeService + + +def _is_lms_request(request): + """Check if request comes from LMS (has a valid signed URL).""" + return request.find_service(SecureLinkService).request_has_valid_token(request) + + +def _restricted_response(request, target_url=None): + """Return the restricted access page response.""" + request.override_renderer = "via:templates/restricted.html.jinja2" + return {"target_url": target_url} def _api_headers(request): @@ -21,6 +32,14 @@ def _api_headers(request): {"url": fields.Url(required=True)}, location="query", unknown=marshmallow.INCLUDE ) def youtube(request, url, **kwargs): + """YouTube video viewer. + + If the request comes through LMS (valid signed URL), serve the viewer. + Otherwise, show the restricted access page. + """ + if not _is_lms_request(request): + return _restricted_response(request, target_url=url) + youtube_service = request.find_service(YouTubeService) if not youtube_service.enabled: @@ -36,7 +55,6 @@ def youtube(request, url, **kwargs): _, client_config = Configuration.extract_from_params(kwargs) return { - # Common video player fields "client_embed_url": request.registry.settings["client_embed_url"], "client_config": client_config, "player": "youtube", @@ -50,7 +68,6 @@ def youtube(request, url, **kwargs): "headers": _api_headers(request), } }, - # Fields specific to YouTube video player "video_id": video_id, } @@ -71,6 +88,15 @@ def youtube(request, url, **kwargs): unknown=marshmallow.INCLUDE, ) def video(request, **kwargs): + """HTML video viewer. + + If the request comes through LMS (valid signed URL), serve the viewer. + Otherwise, show the restricted access page. + """ + if not _is_lms_request(request): + target_url = kwargs.get("url") + return _restricted_response(request, target_url=target_url) + allow_download = kwargs.get("allow_download", True) url = kwargs["url"] media_url = kwargs.get("media_url") @@ -81,7 +107,6 @@ def video(request, **kwargs): _, client_config = Configuration.extract_from_params(kwargs) return { - # Common video player fields "client_embed_url": request.registry.settings["client_embed_url"], "client_config": client_config, "player": "html-video", @@ -97,7 +122,6 @@ def video(request, **kwargs): "headers": _api_headers(request), }, }, - # Fields specific to HTML video player. "allow_download": allow_download, "video_src": media_url, }