diff --git a/CHANGELOG.md b/CHANGELOG.md index 152d6ed0d..0cdcfb9e6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ and this project adheres to - ✨(backend) add a quota_excluded flag on items - ✨(backend) apply per-audience attributes to external api items - ✨(backend) add a grant_unlimited_storage command +- ✨(wopi) verify the WOPI request proof signature ### Changed diff --git a/src/backend/core/api/viewsets.py b/src/backend/core/api/viewsets.py index e09260ab5..6dbf1a323 100644 --- a/src/backend/core/api/viewsets.py +++ b/src/backend/core/api/viewsets.py @@ -1788,7 +1788,7 @@ def wopi(self, request, *args, **kwargs): if request.user.is_authenticated and request.user.language else settings.LANGUAGE_CODE ) - launch_url = compute_wopi_launch_url(wopi_client["url"], get_file_info, language) + launch_url = compute_wopi_launch_url(wopi_client.get("launch_url"), get_file_info, language) return drf.response.Response( { diff --git a/src/backend/core/tests/items/test_api_items_wopi.py b/src/backend/core/tests/items/test_api_items_wopi.py index 2de89ce30..4b68b46b7 100644 --- a/src/backend/core/tests/items/test_api_items_wopi.py +++ b/src/backend/core/tests/items/test_api_items_wopi.py @@ -40,11 +40,14 @@ def configure_wopi_settings(valid_mimetype, valid_wopi_launch_url): { "mimetypes": { valid_mimetype: { - "url": valid_wopi_launch_url, + "launch_url": valid_wopi_launch_url, "client": "vendorA", }, }, "extensions": {}, + "vendorA": { + "proof_keys": {}, + }, }, ) diff --git a/src/backend/wopi/authentication.py b/src/backend/wopi/authentication.py index 69677955f..492e012a1 100644 --- a/src/backend/wopi/authentication.py +++ b/src/backend/wopi/authentication.py @@ -6,28 +6,29 @@ from wopi.services.access import AccessError, AccessUserItemService +def get_access_token(request): + """Look for the access_token in query params first, then headers.""" + access_token = request.query_params.get("access_token") + + if not access_token: + access_token = request.headers.get("Authorization", "") + if access_token.startswith("Bearer "): + access_token = access_token[7:] + + return access_token + + class WopiAccessTokenAuthentication(BaseAuthentication): """ WOPI access token authentication. """ - def _get_access_token(self, request): - """Look for the access_token in query params first, then headers.""" - access_token = request.query_params.get("access_token") - - if not access_token: - access_token = request.headers.get("Authorization", "") - if access_token.startswith("Bearer "): - access_token = access_token[7:] - - return access_token - def authenticate(self, request): """ Authenticate the request. """ # First check if the access token is present in the request - access_token = self._get_access_token(request) + access_token = get_access_token(request) if not access_token: raise AuthenticationFailed("Access token not provided") diff --git a/src/backend/wopi/exceptions.py b/src/backend/wopi/exceptions.py new file mode 100644 index 000000000..0af72e67d --- /dev/null +++ b/src/backend/wopi/exceptions.py @@ -0,0 +1,9 @@ +"""WOPI exceptions module.""" + + +class WopiRequestSignatureError(Exception): + """Exception for when a request signature is invalid.""" + + def __init__(self, message="Invalid request signature"): + self.message = message + super().__init__(self.message) diff --git a/src/backend/wopi/tasks/configure_wopi.py b/src/backend/wopi/tasks/configure_wopi.py index 5c9828902..fce4c4e11 100644 --- a/src/backend/wopi/tasks/configure_wopi.py +++ b/src/backend/wopi/tasks/configure_wopi.py @@ -1,11 +1,15 @@ """Task configuring WOPI using discovery url.""" +from base64 import b64decode + from django.conf import settings from django.core.cache import cache import requests from celery import Celery from celery.schedules import crontab +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicNumbers from defusedxml.ElementTree import fromstring from drive.celery_app import app as celery_app @@ -44,6 +48,19 @@ def configure_wopi_clients(): ) +def build_rsa_public_key(modulus, exponent): + """Build RSA public key from modulus and exponent.""" + mod = int(b64decode(modulus).hex(), 16) + exp = int(b64decode(exponent).hex(), 16) + + rsa_public_key = RSAPublicNumbers(exp, mod).public_key() + + return rsa_public_key.public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ) + + def _configure_wopi_client_from_discovery(client, discovery_url): """Configure wopi client from discovery url.""" @@ -67,6 +84,27 @@ def _configure_wopi_client_from_discovery(client, discovery_url): if net_zone is None: raise RuntimeError(f"net-zone element not found in discovery url for wopi client {client}") + proof_key_node = root.find(".//proof-key") + proof_keys = {} + + if proof_key_node is not None: + # build current and old public key + current_public_key = build_rsa_public_key( + proof_key_node.get("modulus"), proof_key_node.get("exponent") + ) + old_public_key = build_rsa_public_key( + proof_key_node.get("oldmodulus"), proof_key_node.get("oldexponent") + ) + + proof_keys = { + "public_key": current_public_key, + "old_public_key": old_public_key, + } + + wopi_configuration[client] = { + "proof_keys": proof_keys, + } + # Iterate through all app elements for app in net_zone.findall(".//app"): app_name = app.get("name") @@ -85,7 +123,7 @@ def _configure_wopi_client_from_discovery(client, discovery_url): continue wopi_configuration["mimetypes"][mimetype] = { - "url": action.get("urlsrc"), + "launch_url": action.get("urlsrc"), "client": client, } @@ -96,7 +134,7 @@ def _configure_wopi_client_from_discovery(client, discovery_url): continue wopi_configuration["extensions"][extension] = { - "url": action.get("urlsrc"), + "launch_url": action.get("urlsrc"), "client": client, } diff --git a/src/backend/wopi/tests/conftest.py b/src/backend/wopi/tests/conftest.py index 50aaba8e5..c1d011e08 100644 --- a/src/backend/wopi/tests/conftest.py +++ b/src/backend/wopi/tests/conftest.py @@ -4,9 +4,38 @@ import pytest +from wopi.tasks.configure_wopi import WOPI_CONFIGURATION_CACHE_KEY + @pytest.fixture(autouse=True) def clear_cache(): """Fixture to clear the cache before each test.""" yield cache.clear() + + +@pytest.fixture +def configure_wopi_clients(): + """Configure wopi clients.""" + + wopi_configuration = { + "mimetypes": { + "text/plain": { + "launch_url": "http://localhost:9980/browser/0968141f2c/cool.html?", + "client": "vendorA", + } + }, + "extensions": { + "txt": { + "launch_url": "http://localhost:9980/browser/0968141f2c/cool.html?", + "client": "vendorA", + } + }, + "vendorA": { + "proof_keys": {"public_key": b"public_proof_key\n"}, + }, + } + cache.set(WOPI_CONFIGURATION_CACHE_KEY, wopi_configuration) + + yield wopi_configuration + cache.delete(WOPI_CONFIGURATION_CACHE_KEY) diff --git a/src/backend/wopi/tests/tasks/test_configure_wopi.py b/src/backend/wopi/tests/tasks/test_configure_wopi.py index b94631850..a48a36571 100644 --- a/src/backend/wopi/tests/tasks/test_configure_wopi.py +++ b/src/backend/wopi/tests/tasks/test_configure_wopi.py @@ -49,16 +49,83 @@ def test_configure_wopi_clients(settings): assert cache.get(WOPI_CONFIGURATION_CACHE_KEY) == { "mimetypes": { "application/vnd.oasis.opendocument.text": { - "url": "http://localhost:9980/browser/0968141f2c/cool.html?", + "launch_url": "http://localhost:9980/browser/0968141f2c/cool.html?", "client": "vendorA", }, }, "extensions": { "odt": { - "url": "http://localhost:9980/browser/0968141f2c/cool.html?", + "launch_url": "http://localhost:9980/browser/0968141f2c/cool.html?", "client": "vendorA", }, }, + "vendorA": { + "proof_keys": {}, + }, + } + + +@responses.activate +def test_configure_wopi_clients_with_proof_keys(settings): + """Test the configure_wopi celery task with client using proof key.""" + + settings.WOPI_CLIENTS = ["vendorA"] + settings.WOPI_CLIENTS_CONFIGURATION = { + "vendorA": { + "discovery_url": "https://vendorA.com/hosting/discovery", + } + } + + # pylint: disable=line-too-long + responses.add( + responses.GET, + "https://vendorA.com/hosting/discovery", + body=""" + + + + + + + + + + + + +""", + ) + + assert cache.get(WOPI_CONFIGURATION_CACHE_KEY) is None + + configure_wopi_clients() + + # pylint: disable=line-too-long + assert cache.get(WOPI_CONFIGURATION_CACHE_KEY) == { + "mimetypes": { + "application/vnd.oasis.opendocument.text": { + "launch_url": "http://localhost:9980/browser/0968141f2c/cool.html?", + "client": "vendorA", + }, + }, + "extensions": { + "odt": { + "launch_url": "http://localhost:9980/browser/0968141f2c/cool.html?", + "client": "vendorA", + }, + }, + "vendorA": { + "proof_keys": { + "old_public_key": b"-----BEGIN PUBLIC KEY-----\nMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA04SCCYR0PEvAAVGxNyua\neYwNQ9gOsnbQaC4jQF0CEPSkW9Aj/y8y6rwEoHRBJmGeoiemd+0XEWi1LPFeCo+7\nC98VsPqWS4kPPx4mVJWTGSR8Do9/CMm4Ezs/bxj2Y/aNIuEwMOdVTSlR8woDt73D\nkwOTDbpSOrOk5ras63yCNu4tLk5xexW6I8vkykRXjJtU59BCzrQzBwwSAlHUTkiZ\nyXW8tqtELEkRoMWMSaOE5y89v+SsJbmNpKqI8uIKXKgaY6vtjYpS+TjcED5M40J6\nJDjy50uc8zXTAy0UqUI+G5hho70fKSTgs4D+oljG2T1amUoFNyKCnVW8q1O6IeLm\n+wIDAQAB\n-----END PUBLIC KEY-----\n", + "public_key": b"-----BEGIN PUBLIC KEY-----\nMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA04SCCYR0PEvAAVGxNyua\neYwNQ9gOsnbQaC4jQF0CEPSkW9Aj/y8y6rwEoHRBJmGeoiemd+0XEWi1LPFeCo+7\nC98VsPqWS4kPPx4mVJWTGSR8Do9/CMm4Ezs/bxj2Y/aNIuEwMOdVTSlR8woDt73D\nkwOTDbpSOrOk5ras63yCNu4tLk5xexW6I8vkykRXjJtU59BCzrQzBwwSAlHUTkiZ\nyXW8tqtELEkRoMWMSaOE5y89v+SsJbmNpKqI8uIKXKgaY6vtjYpS+TjcED5M40J6\nJDjy50uc8zXTAy0UqUI+G5hho70fKSTgs4D+oljG2T1amUoFNyKCnVW8q1O6IeLm\n+wIDAQAB\n-----END PUBLIC KEY-----\n", + }, + }, } diff --git a/src/backend/wopi/tests/utils/__init__.py b/src/backend/wopi/tests/utils/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/backend/wopi/tests/utils/test_signature.py b/src/backend/wopi/tests/utils/test_signature.py new file mode 100644 index 000000000..d54559cad --- /dev/null +++ b/src/backend/wopi/tests/utils/test_signature.py @@ -0,0 +1,349 @@ +"""Tests for the signature utils.""" + +from base64 import b64encode +from datetime import datetime, timezone + +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import padding, rsa + +from wopi.utils import signature + +# ---------- TESTS FOR ticks_to_datetime ---------- + + +def test_ticks_to_datetime(): + """Test the ticks_to_datetime function.""" + # Test epoch (.NET ticks = 0) + assert signature.ticks_to_datetime(0) == datetime(1, 1, 1, 0, 0, 0, 0, timezone.utc) + + # Test Unix epoch (.NET ticks for 1970-01-01) + assert signature.ticks_to_datetime(621355968000000000) == datetime( + 1970, 1, 1, 0, 0, 0, 0, timezone.utc + ) + + # Test a specific date (2024-01-01 00:00:00 UTC) + # Calculate ticks for 2024-01-01 + expected_2024 = datetime(2024, 1, 1, 0, 0, 0, 0, timezone.utc) + unix_timestamp_2024 = expected_2024.timestamp() + dotnet_epoch_ticks = 621355968000000000 + ticks_per_second = 10000000 + ticks_2024 = int(dotnet_epoch_ticks + (unix_timestamp_2024 * ticks_per_second)) + assert signature.ticks_to_datetime(ticks_2024) == expected_2024 + + # Test with fractional seconds + # Add 1234567 ticks = 123456700 nanoseconds = 123.4567 milliseconds + ticks_with_fraction = 621355968000000000 + 1234567 + result = signature.ticks_to_datetime(ticks_with_fraction) + assert result.year == 1970 + assert result.month == 1 + assert result.day == 1 + # 1234567 ticks = 123456700 nanoseconds = 123456.7 microseconds + # Round to nearest microsecond = 123457 microseconds + assert result.microsecond == 123457 + + +# ---------- TESTS FOR build_expected_proof ---------- + + +def test_build_expected_proof(): + """Test the build_expected_proof function.""" + access_token = "test_token" + url = "https://example.com/wopi/files/123" + timestamp = 621355968000000000 + + result = signature.build_expected_proof(access_token, url, timestamp) + + # Verify structure: [len(token)] [token] [len(url)] [url] [len(timestamp)] [timestamp] + token_bytes = access_token.encode("utf-8") + url_bytes = url.upper().encode("utf-8") # URL is uppercased + timestamp_bytes = signature.encode_timestamp(timestamp) + + expected = ( + signature.encode_number(len(token_bytes)) + + token_bytes + + signature.encode_number(len(url_bytes)) + + url_bytes + + signature.encode_number(len(timestamp_bytes)) + + timestamp_bytes + ) + + assert result == expected + + +def test_build_expected_proof_url_uppercase(): + """Test that build_expected_proof uppercases the URL.""" + access_token = "token" + url_lower = "https://example.com/test" + url_upper = "HTTPS://EXAMPLE.COM/TEST" + timestamp = 1000000 + + result_lower = signature.build_expected_proof(access_token, url_lower, timestamp) + result_upper = signature.build_expected_proof(access_token, url_upper, timestamp) + + # Both should produce the same result + assert result_lower == result_upper + + +def test_build_expected_proof_empty_strings(): + """Test build_expected_proof with empty strings.""" + access_token = "" + url = "" + timestamp = 0 + + result = signature.build_expected_proof(access_token, url, timestamp) + + # Should still have proper structure with length prefixes + assert len(result) == 4 + 0 + 4 + 0 + 4 + 8 # Three length prefixes + timestamp + + +def test_build_expected_proof_unicode(): + """Test build_expected_proof with unicode characters.""" + access_token = "tëst_tökën" + url = "https://example.com/test/path" + timestamp = 621355968000000000 + + result = signature.build_expected_proof(access_token, url, timestamp) + + # Should handle unicode correctly + assert len(result) > 0 + assert access_token.encode("utf-8") in result + + +# ---------- TESTS FOR verify_wopi_proof ---------- + + +def _generate_rsa_keypair(): + """Generate an RSA key pair for testing.""" + private_key = rsa.generate_private_key( + public_exponent=65537, + key_size=2048, + ) + public_key = private_key.public_key() + return private_key, public_key + + +def _serialize_public_key(public_key): + """Serialize RSA public key to PEM format.""" + return public_key.public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ) + + +def _sign_data(private_key, data): + """Sign data with RSA private key using PKCS#1 v1.5 padding and SHA256.""" + signature_bytes = private_key.sign(data, padding.PKCS1v15(), hashes.SHA256()) + return b64encode(signature_bytes).decode("utf-8") + + +def test_verify_wopi_proof_valid_signature(): + """Test verify_wopi_proof with a valid signature.""" + private_key, public_key = _generate_rsa_keypair() + public_key_pem = _serialize_public_key(public_key) + + access_token = "test_token" + url = "https://example.com/wopi/files/123" + timestamp = 621355968000000000 + expected_proof = signature.build_expected_proof(access_token, url, timestamp) + + signature_value = _sign_data(private_key, expected_proof) + + proof_keys = {"public_key": public_key_pem} + + result = signature.verify_wopi_proof( + proof_keys=proof_keys, + signature=signature_value, + signature_old=None, + expected_proof=expected_proof, + ) + + assert result is True + + +def test_verify_wopi_proof_invalid_signature(): + """Test verify_wopi_proof with an invalid signature.""" + private_key, public_key = _generate_rsa_keypair() + public_key_pem = _serialize_public_key(public_key) + + access_token = "test_token" + url = "https://example.com/wopi/files/123" + timestamp = 621355968000000000 + expected_proof = signature.build_expected_proof(access_token, url, timestamp) + + # Create invalid signature (sign different data) + wrong_proof = signature.build_expected_proof("wrong_token", url, timestamp) + signature_value = _sign_data(private_key, wrong_proof) + + proof_keys = {"public_key": public_key_pem} + + result = signature.verify_wopi_proof( + proof_keys=proof_keys, + signature=signature_value, + signature_old=None, + expected_proof=expected_proof, + ) + + assert result is False + + +def test_verify_wopi_proof_with_signature_old_valid(): + """Test verify_wopi_proof when signature_old is valid with current key.""" + private_key, public_key = _generate_rsa_keypair() + public_key_pem = _serialize_public_key(public_key) + + access_token = "test_token" + url = "https://example.com/wopi/files/123" + timestamp = 621355968000000000 + expected_proof = signature.build_expected_proof(access_token, url, timestamp) + + # Invalid current signature + invalid_signature = b64encode(b"invalid_signature").decode("utf-8") + + # Valid old signature + signature_old = _sign_data(private_key, expected_proof) + + proof_keys = {"public_key": public_key_pem} + + result = signature.verify_wopi_proof( + proof_keys=proof_keys, + signature=invalid_signature, + signature_old=signature_old, + expected_proof=expected_proof, + ) + + assert result is True + + +def test_verify_wopi_proof_with_old_public_key(): + """Test verify_wopi_proof when signature is valid with old public key.""" + # Generate two key pairs (current and old) + old_private_key, old_public_key = _generate_rsa_keypair() + _, current_public_key = _generate_rsa_keypair() + + old_public_key_pem = _serialize_public_key(old_public_key) + current_public_key_pem = _serialize_public_key(current_public_key) + + access_token = "test_token" + url = "https://example.com/wopi/files/123" + timestamp = 621355968000000000 + expected_proof = signature.build_expected_proof(access_token, url, timestamp) + + # Sign with old private key (simulating key rotation scenario) + signature_value = _sign_data(old_private_key, expected_proof) + + proof_keys = { + "public_key": current_public_key_pem, + "old_public_key": old_public_key_pem, + } + + result = signature.verify_wopi_proof( + proof_keys=proof_keys, + signature=signature_value, + signature_old=None, + expected_proof=expected_proof, + ) + + assert result is True + + +def test_verify_wopi_proof_all_invalid(): + """Test verify_wopi_proof when all signatures are invalid.""" + _, public_key = _generate_rsa_keypair() + _, old_public_key = _generate_rsa_keypair() + public_key_pem = _serialize_public_key(public_key) + old_public_key_pem = _serialize_public_key(old_public_key) + + access_token = "test_token" + url = "https://example.com/wopi/files/123" + timestamp = 621355968000000000 + expected_proof = signature.build_expected_proof(access_token, url, timestamp) + + # All invalid signatures + invalid_signature = b64encode(b"invalid_signature").decode("utf-8") + invalid_signature_old = b64encode(b"invalid_signature_old").decode("utf-8") + + proof_keys = { + "public_key": public_key_pem, + "old_public_key": old_public_key_pem, + } + + result = signature.verify_wopi_proof( + proof_keys=proof_keys, + signature=invalid_signature, + signature_old=invalid_signature_old, + expected_proof=expected_proof, + ) + + assert result is False + + +def test_verify_wopi_proof_malformed_signature(): + """Test verify_wopi_proof with a malformed base64 signature.""" + _, public_key = _generate_rsa_keypair() + public_key_pem = _serialize_public_key(public_key) + + access_token = "test_token" + url = "https://example.com/wopi/files/123" + timestamp = 621355968000000000 + expected_proof = signature.build_expected_proof(access_token, url, timestamp) + + proof_keys = {"public_key": public_key_pem} + + result = signature.verify_wopi_proof( + proof_keys=proof_keys, + signature="malformed base64!", + signature_old=None, + expected_proof=expected_proof, + ) + + assert result is False + + +def test_verify_wopi_proof_malformed_signature_old(): + """Test verify_wopi_proof with a malformed base64 old signature.""" + _, public_key = _generate_rsa_keypair() + public_key_pem = _serialize_public_key(public_key) + + access_token = "test_token" + url = "https://example.com/wopi/files/123" + timestamp = 621355968000000000 + expected_proof = signature.build_expected_proof(access_token, url, timestamp) + + invalid_signature = b64encode(b"invalid_signature").decode("utf-8") + + proof_keys = {"public_key": public_key_pem} + + result = signature.verify_wopi_proof( + proof_keys=proof_keys, + signature=invalid_signature, + signature_old="malformed base64!", + expected_proof=expected_proof, + ) + + assert result is False + + +def test_verify_wopi_proof_wrong_key(): + """Test verify_wopi_proof when signature is signed with a different key.""" + # Generate two different key pairs + private_key1, _ = _generate_rsa_keypair() + _, public_key2 = _generate_rsa_keypair() + + # Sign with key1 but verify with key2 + access_token = "test_token" + url = "https://example.com/wopi/files/123" + timestamp = 621355968000000000 + expected_proof = signature.build_expected_proof(access_token, url, timestamp) + + signature_value = _sign_data(private_key1, expected_proof) + + proof_keys = {"public_key": _serialize_public_key(public_key2)} + + result = signature.verify_wopi_proof( + proof_keys=proof_keys, + signature=signature_value, + signature_old=None, + expected_proof=expected_proof, + ) + + assert result is False diff --git a/src/backend/wopi/tests/viewset/test_check_file_info.py b/src/backend/wopi/tests/viewset/test_check_file_info.py index f5c74f8c5..993fc668f 100644 --- a/src/backend/wopi/tests/viewset/test_check_file_info.py +++ b/src/backend/wopi/tests/viewset/test_check_file_info.py @@ -1,17 +1,22 @@ """Testing the check file info endpoint""" +from datetime import timedelta from io import BytesIO +from unittest import mock from django.contrib.auth.models import AnonymousUser from django.core.cache import cache from django.core.files.storage import default_storage +from django.utils.timezone import now import pytest from rest_framework.test import APIClient from core import factories, models +from wopi.exceptions import WopiRequestSignatureError from wopi.services.access import AccessUserItemService from wopi.tasks.configure_wopi import WOPI_CONFIGURATION_CACHE_KEY +from wopi.utils import signature as signature_utils from drive.settings import Base @@ -278,6 +283,7 @@ def test_check_file_info_supports_rename_override(settings, monkeypatch): "client": "collabora", }, }, + "collabora": {"proof_keys": {}}, }, ) @@ -304,3 +310,244 @@ def test_check_file_info_non_existing_access_token(): HTTP_AUTHORIZATION="Bearer not_existing_token", ) assert response.status_code == 403 + + +def test_check_file_info_connected_user_with_access_with_valid_signature( + configure_wopi_clients, +): + """Check signature validation for a connected user with access.""" + wopi_configuration = configure_wopi_clients + folder = factories.ItemFactory( + type=models.ItemTypeChoices.FOLDER, + ) + item = factories.ItemFactory( + parent=folder, + type=models.ItemTypeChoices.FILE, + filename="wopi_test.txt", + update_upload_state=models.ItemUploadStateChoices.READY, + link_reach=models.LinkReachChoices.RESTRICTED, + link_role=models.LinkRoleChoices.EDITOR, + mimetype="text/plain", + ) + user = factories.UserFactory() + factories.UserItemAccessFactory(item=item, user=user, role=models.RoleChoices.EDITOR) + + default_storage.connection.meta.client.put_object( + Bucket=default_storage.bucket_name, + Key=item.file_key, + Body=BytesIO(b"my prose"), + ContentType="text/plain", + ) + + service = AccessUserItemService() + access_token, _ = service.insert_new_access(item, user) + + wopi_timestamp = ( + signature_utils.DOTNET_EPOCH_TICKS + now().timestamp() * signature_utils.TICKS_PER_SECOND + ) + + client = APIClient() + with mock.patch.object(signature_utils, "verify_wopi_proof") as mock_verify_wopi_proof: + mock_verify_wopi_proof.return_value = True + response = client.get( + f"/api/v1.0/wopi/files/{item.id}/", + HTTP_AUTHORIZATION=f"Bearer {access_token}", + HTTP_X_WOPI_PROOF="valid_signature", + HTTP_X_WOPI_TIMESTAMP=wopi_timestamp, + ) + + assert response.status_code == 200 + + mock_verify_wopi_proof.assert_called_once_with( + wopi_configuration["vendorA"]["proof_keys"], + "valid_signature", + None, + mock.ANY, + ) + + +def test_check_file_info_connected_user_with_access_with_invalid_signature( + configure_wopi_clients, +): + """Check signature validation for a connected user with access.""" + wopi_configuration = configure_wopi_clients + folder = factories.ItemFactory( + type=models.ItemTypeChoices.FOLDER, + ) + item = factories.ItemFactory( + parent=folder, + type=models.ItemTypeChoices.FILE, + filename="wopi_test.txt", + update_upload_state=models.ItemUploadStateChoices.READY, + link_reach=models.LinkReachChoices.RESTRICTED, + link_role=models.LinkRoleChoices.EDITOR, + mimetype="text/plain", + ) + user = factories.UserFactory() + factories.UserItemAccessFactory(item=item, user=user, role=models.RoleChoices.EDITOR) + + default_storage.connection.meta.client.put_object( + Bucket=default_storage.bucket_name, + Key=item.file_key, + Body=BytesIO(b"my prose"), + ContentType="text/plain", + ) + + service = AccessUserItemService() + access_token, _ = service.insert_new_access(item, user) + + wopi_timestamp = ( + signature_utils.DOTNET_EPOCH_TICKS + now().timestamp() * signature_utils.TICKS_PER_SECOND + ) + + client = APIClient() + with mock.patch.object(signature_utils, "verify_wopi_proof") as mock_verify_wopi_proof: + mock_verify_wopi_proof.return_value = False + with pytest.raises(WopiRequestSignatureError, match="Invalid request signature"): + client.get( + f"/api/v1.0/wopi/files/{item.id}/", + HTTP_AUTHORIZATION=f"Bearer {access_token}", + HTTP_X_WOPI_PROOF="invalid_signature", + HTTP_X_WOPI_TIMESTAMP=wopi_timestamp, + ) + + mock_verify_wopi_proof.assert_called_once_with( + wopi_configuration["vendorA"]["proof_keys"], + "invalid_signature", + None, + mock.ANY, + ) + + +def test_check_file_info_connected_user_with_access_with_expired_wopi_timestamp( + configure_wopi_clients, # pylint: disable=unused-argument +): + """Check signature validation for a connected user with access with expired wopi timestamp.""" + folder = factories.ItemFactory( + type=models.ItemTypeChoices.FOLDER, + ) + item = factories.ItemFactory( + parent=folder, + type=models.ItemTypeChoices.FILE, + filename="wopi_test.txt", + update_upload_state=models.ItemUploadStateChoices.READY, + link_reach=models.LinkReachChoices.RESTRICTED, + link_role=models.LinkRoleChoices.EDITOR, + mimetype="text/plain", + ) + user = factories.UserFactory() + factories.UserItemAccessFactory(item=item, user=user, role=models.RoleChoices.EDITOR) + + default_storage.connection.meta.client.put_object( + Bucket=default_storage.bucket_name, + Key=item.file_key, + Body=BytesIO(b"my prose"), + ContentType="text/plain", + ) + + service = AccessUserItemService() + access_token, _ = service.insert_new_access(item, user) + + wopi_timestamp = ( + signature_utils.DOTNET_EPOCH_TICKS + + (now() - timedelta(minutes=20)).timestamp() * signature_utils.TICKS_PER_SECOND + ) + + client = APIClient() + with mock.patch.object(signature_utils, "verify_wopi_proof") as mock_verify_wopi_proof: + with pytest.raises( + WopiRequestSignatureError, match="Timestamp is too old, request rejected" + ): + client.get( + f"/api/v1.0/wopi/files/{item.id}/", + HTTP_AUTHORIZATION=f"Bearer {access_token}", + HTTP_X_WOPI_PROOF="invalid_signature", + HTTP_X_WOPI_TIMESTAMP=wopi_timestamp, + ) + + mock_verify_wopi_proof.assert_not_called() + + +def test_check_file_info_connected_user_with_access_proof_keys_configured_but_no_signature_provided( + configure_wopi_clients, # pylint: disable=unused-argument +): + """Check signature validation for a connected user with access with no signature provided.""" + folder = factories.ItemFactory( + type=models.ItemTypeChoices.FOLDER, + ) + item = factories.ItemFactory( + parent=folder, + type=models.ItemTypeChoices.FILE, + filename="wopi_test.txt", + update_upload_state=models.ItemUploadStateChoices.READY, + link_reach=models.LinkReachChoices.RESTRICTED, + link_role=models.LinkRoleChoices.EDITOR, + mimetype="text/plain", + ) + user = factories.UserFactory() + factories.UserItemAccessFactory(item=item, user=user, role=models.RoleChoices.EDITOR) + + default_storage.connection.meta.client.put_object( + Bucket=default_storage.bucket_name, + Key=item.file_key, + Body=BytesIO(b"my prose"), + ContentType="text/plain", + ) + + service = AccessUserItemService() + access_token, _ = service.insert_new_access(item, user) + + client = APIClient() + with mock.patch.object(signature_utils, "verify_wopi_proof") as mock_verify_wopi_proof: + with pytest.raises( + WopiRequestSignatureError, match="No signature provided, request rejected" + ): + client.get( + f"/api/v1.0/wopi/files/{item.id}/", + HTTP_AUTHORIZATION=f"Bearer {access_token}", + ) + + mock_verify_wopi_proof.assert_not_called() + + +def test_check_file_info_connected_user_with_access_proof_keys_configured_but_no_timestamp_provided( + configure_wopi_clients, # pylint: disable=unused-argument +): + """Check signature validation for a connected user with access with no timestamp provided.""" + folder = factories.ItemFactory( + type=models.ItemTypeChoices.FOLDER, + ) + item = factories.ItemFactory( + parent=folder, + type=models.ItemTypeChoices.FILE, + filename="wopi_test.txt", + update_upload_state=models.ItemUploadStateChoices.READY, + link_reach=models.LinkReachChoices.RESTRICTED, + link_role=models.LinkRoleChoices.EDITOR, + mimetype="text/plain", + ) + user = factories.UserFactory() + factories.UserItemAccessFactory(item=item, user=user, role=models.RoleChoices.EDITOR) + + default_storage.connection.meta.client.put_object( + Bucket=default_storage.bucket_name, + Key=item.file_key, + Body=BytesIO(b"my prose"), + ContentType="text/plain", + ) + + service = AccessUserItemService() + access_token, _ = service.insert_new_access(item, user) + + client = APIClient() + with mock.patch.object(signature_utils, "verify_wopi_proof") as mock_verify_wopi_proof: + with pytest.raises( + WopiRequestSignatureError, match="No timestamp provided, request rejected" + ): + client.get( + f"/api/v1.0/wopi/files/{item.id}/", + HTTP_AUTHORIZATION=f"Bearer {access_token}", + HTTP_X_WOPI_PROOF="invalid_signature", + ) + + mock_verify_wopi_proof.assert_not_called() diff --git a/src/backend/wopi/utils/__init__.py b/src/backend/wopi/utils/__init__.py index bf2834b09..1dadefe2b 100644 --- a/src/backend/wopi/utils/__init__.py +++ b/src/backend/wopi/utils/__init__.py @@ -23,7 +23,7 @@ def is_item_wopi_supported(item, user): def get_wopi_client_config(item, user): - """make + """ Get the WOPI client configuration for an item. """ if ( @@ -33,7 +33,7 @@ def get_wopi_client_config(item, user): ): return None - wopi_configuration = cache.get(WOPI_CONFIGURATION_CACHE_KEY, default=WOPI_DEFAULT_CONFIGURATION) + wopi_configuration = get_wopi_configuration() if not wopi_configuration: return None @@ -50,6 +50,23 @@ def get_wopi_client_config(item, user): return result +def get_wopi_client_proof_keys(item, user): + """get the wopi proof keys for an item""" + wopi_client_config = get_wopi_client_config(item, user) + + if not wopi_client_config: + return None + + wopi_configuration = get_wopi_configuration() + + return wopi_configuration[wopi_client_config["client"]]["proof_keys"] + + +def get_wopi_configuration(): + """get the wopi configuration""" + return cache.get(WOPI_CONFIGURATION_CACHE_KEY, default=WOPI_DEFAULT_CONFIGURATION) + + def compute_wopi_launch_url(launch_url, get_file_info_path, lang=None): """ Compute the WOPI launch URL for an item. diff --git a/src/backend/wopi/utils/signature.py b/src/backend/wopi/utils/signature.py new file mode 100644 index 000000000..dfdd43249 --- /dev/null +++ b/src/backend/wopi/utils/signature.py @@ -0,0 +1,152 @@ +""" +Utils for validating WOPI proof signatures. + +https://learn.microsoft.com/en-us/microsoft-365/cloud-storage-partner-program/online/scenarios/proofkey +""" + +import struct +from base64 import b64decode +from datetime import datetime, timezone + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import padding + +# ---------- HELPERS ---------- + +# .NET DateTime epoch: January 1, 0001 00:00:00 +# Unix datetime epoch: January 1, 1970 00:00:00 UTC +# Ticks between these epochs: 621,355,968,000,000,000 +DOTNET_EPOCH_TICKS = 621355968000000000 + +# Each tick is 100 nanoseconds = 0.0000001 seconds +TICKS_PER_SECOND = 10000000 + + +def ticks_to_datetime(ticks: int) -> datetime: + """Convert a .NET DateTime ticks to a datetime object. + + .NET DateTime ticks represent the number of 100-nanosecond intervals + since January 1, 0001 00:00:00.000 in the Gregorian calendar. + + Args: + ticks: Integer representing .NET DateTime ticks + + Returns: + datetime: Python datetime object corresponding to the ticks + """ + + # Convert .NET ticks to Unix timestamp (seconds since 1970-01-01) + unix_seconds = (ticks - DOTNET_EPOCH_TICKS) / TICKS_PER_SECOND + + # Create datetime from Unix timestamp (UTC) + return datetime.fromtimestamp(unix_seconds, timezone.utc) + + +# ---------- BYTE ENCODING (C# equivalent) ---------- + + +def encode_number(n: int) -> bytes: + """Encode a number as a 4-byte big endian.""" + return struct.pack(">I", n) + + +def encode_timestamp(ts: int) -> bytes: + """ + Encode a timestamp as a 8-byte big endian. + A timestamp from the wopi header is a .NET DateTime ticks + https://learn.microsoft.com/fr-fr/dotnet/api/system.datetime.ticks + """ + return struct.pack(">Q", ts) + + +# ---------- BUILD EXPECTED PROOF ---------- + + +def build_expected_proof(access_token: str, url: str, timestamp: int) -> bytes: + """ + Build the expected proof for a WOPI request. + + https://learn.microsoft.com/en-us/microsoft-365/cloud-storage-partner-program/online/scenarios/proofkeys#constructing-the-expected-proof + """ + access_token_bytes = access_token.encode("utf-8") + url_bytes = url.upper().encode("utf-8") + timestamp_bytes = encode_timestamp(timestamp) + + proof = bytearray() + proof.extend(encode_number(len(access_token_bytes))) + proof.extend(access_token_bytes) + proof.extend(encode_number(len(url_bytes))) + proof.extend(url_bytes) + proof.extend(encode_number(len(timestamp_bytes))) + proof.extend(timestamp_bytes) + + return bytes(proof) + + +# ---------- RSA SIGNATURE VERIFICATION ---------- + + +def verify_wopi_proof( + proof_keys: dict[str, bytes], + signature: str, + signature_old: str | None, + expected_proof: bytes, +) -> bool: + """ + Verify the RSA-SHA256 signature for a WOPI singature using PKCS#1 v1.5 padding. + + https://learn.microsoft.com/en-us/microsoft-365/cloud-storage-partner-program/online/scenarios/proofkeys#verifying-the-proof-keys + """ + + public_key = serialization.load_pem_public_key(proof_keys["public_key"]) + old_public_key = None + if proof_keys.get("old_public_key"): + old_public_key = serialization.load_pem_public_key(proof_keys["old_public_key"]) + + try: + signed_proof = b64decode(signature) + except ValueError: + # A malformed base64 signature cannot be valid + return False + + # Frist try if the current signature is valid with the current public key + try: + public_key.verify( + signed_proof, + expected_proof, + padding.PKCS1v15(), + hashes.SHA256(), + ) + return True + except InvalidSignature: + pass + + # Then try if the signature old value is valid using the current public key + if signature_old: + try: + signed_proof_old = b64decode(signature_old) + public_key.verify( + signed_proof_old, + expected_proof, + padding.PKCS1v15(), + hashes.SHA256(), + ) + return True + except (ValueError, InvalidSignature): + pass + + # Finally try he X-WOPI-Proof value using the old public key + if old_public_key: + try: + old_public_key.verify( + signed_proof, + expected_proof, + padding.PKCS1v15(), + hashes.SHA256(), + ) + return True + except InvalidSignature: + pass + + return False diff --git a/src/backend/wopi/viewsets.py b/src/backend/wopi/viewsets.py index aba32f88a..90661a7c1 100644 --- a/src/backend/wopi/viewsets.py +++ b/src/backend/wopi/viewsets.py @@ -2,6 +2,7 @@ import logging import uuid +from datetime import timedelta from os.path import splitext from django.conf import settings @@ -10,6 +11,7 @@ from django.core.files.storage import default_storage from django.db import transaction from django.http import StreamingHttpResponse +from django.utils.timezone import now from lasuite.malware_detection import malware_detection from rest_framework import viewsets @@ -19,10 +21,16 @@ from core.api.utils import get_item_file_head_object from core.models import Item -from wopi.authentication import WopiAccessTokenAuthentication +from wopi.authentication import WopiAccessTokenAuthentication, get_access_token +from wopi.exceptions import WopiRequestSignatureError from wopi.permissions import AccessTokenPermission from wopi.services.lock import LockService -from wopi.utils import get_wopi_client_config, get_wopi_item_version +from wopi.utils import ( + get_wopi_client_config, + get_wopi_client_proof_keys, + get_wopi_item_version, + signature, +) logger = logging.getLogger(__name__) @@ -30,6 +38,9 @@ HTTP_X_WOPI_LOCK = "HTTP_X_WOPI_LOCK" HTTP_X_WOPI_OLD_LOCK = "HTTP_X_WOPI_OLDLOCK" HTTP_X_WOPI_OVERRIDE = "HTTP_X_WOPI_OVERRIDE" +HTTP_X_WOPI_TIMESTAMP = "HTTP_X_WOPI_TIMESTAMP" +HTTP_X_WOPI_PROOF = "HTTP_X_WOPI_PROOF" +HTTP_X_WOPI_PROOFOLD = "HTTP_X_WOPI_PROOFOLD" X_WOPI_INVALIDFILENAMERROR = "X-WOPI-InvalidFileNameError" X_WOPI_ITEMVERSION = "X-WOPI-ItemVersion" @@ -60,6 +71,48 @@ def get_file_id(self): """Get the file id from the URL path.""" return uuid.UUID(self.kwargs.get("pk")) + def _verify_request_signature(self, request): + """Verify the request signature.""" + proof_keys = get_wopi_client_proof_keys(request.auth.item, request.user) + + if not proof_keys: + # The proof key is not provided by the wopi client, + # so we can't verify the request signature and the request is accepted + return + + request_signature = request.META.get(HTTP_X_WOPI_PROOF) + request_signature_old = request.META.get(HTTP_X_WOPI_PROOFOLD) + + if not request_signature: + raise WopiRequestSignatureError("No signature provided, request rejected") + + string_timestamp = request.META.get(HTTP_X_WOPI_TIMESTAMP) + if not string_timestamp: + raise WopiRequestSignatureError("No timestamp provided, request rejected") + try: + timestamp = int(string_timestamp) + except ValueError as e: + raise WopiRequestSignatureError("Invalid timestamp provided") from e + + datetime_timestamp = signature.ticks_to_datetime(timestamp) + if datetime_timestamp < now() - timedelta(minutes=20): + raise WopiRequestSignatureError("Timestamp is too old, request rejected") + + access_token = get_access_token(request) + expected_proof = signature.build_expected_proof( + access_token, + request.build_absolute_uri(), + timestamp, + ) + + if not signature.verify_wopi_proof( + proof_keys, + request_signature, + request_signature_old, + expected_proof, + ): + raise WopiRequestSignatureError("Invalid request signature") + # pylint: disable=unused-argument def retrieve(self, request, pk=None): """ @@ -69,6 +122,8 @@ def retrieve(self, request, pk=None): item = request.auth.item abilities = item.get_abilities(request.user) + self._verify_request_signature(request) + head_object = get_item_file_head_object(item) wopi_client = get_wopi_client_config(item, request.user) client_options = {} @@ -111,6 +166,7 @@ def file_content(self, request, pk=None): """ Operations to get or put the file content. """ + self._verify_request_signature(request) if request.method == "GET": return self._get_file_content(request, pk) if request.method == "POST": @@ -214,6 +270,9 @@ def detail_post(self, request, pk=None): """ if not request.META.get(HTTP_X_WOPI_OVERRIDE) in self.detail_post_actions: return Response(status=404) + + self._verify_request_signature(request) + item = request.auth.item abilities = item.get_abilities(request.user)