diff --git a/src/matrix_common/types/__init__.py b/src/matrix_common/types/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/matrix_common/types/mxc_uri.py b/src/matrix_common/types/mxc_uri.py new file mode 100644 index 0000000..894a8f7 --- /dev/null +++ b/src/matrix_common/types/mxc_uri.py @@ -0,0 +1,86 @@ +# Copyright 2022 The Matrix.org Foundation C.I.C. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from typing import Type, TypeVar +from urllib.parse import urlparse + +import attr + +MU = TypeVar("MU", bound="MXCUri") + + +@attr.s(frozen=True, slots=True, auto_attribs=True) +class MXCUri: + """Represents a URI that points to a media resource in matrix. + + MXC URIs take the form 'mxc://server_name/media_id'. + """ + + server_name: str + media_id: str + + @classmethod + def from_str(cls: Type[MU], mxc_uri_str: str) -> MU: + """ + Given a str in the form "mxc:///", return an equivalent MXCUri. + + Args: + mxc_uri_str: The MXC Uri as a str. + + Returns: + An MXCUri object with matching attributes. + + Raises: + ValueError: If the str was not a valid MXC Uri. + """ + # Attempt to parse the given URI. This will raise a ValueError if the uri is + # particularly malformed. + parsed_mxc_uri = urlparse(mxc_uri_str) + + # MXC Uri's are pretty bare bones. The scheme must be "mxc", and we don't allow + # any fragments, query parameters or other features. + if ( + # The scheme must be "mxc". + parsed_mxc_uri.scheme != "mxc" + # There must be a host and path provided. + or not parsed_mxc_uri.netloc + or not parsed_mxc_uri.path + or not parsed_mxc_uri.path.startswith("/") + or len(parsed_mxc_uri.path) == 1 # if the path is only '/', aka no Media ID + # There cannot be any fragments, queries or parameters. + or parsed_mxc_uri.fragment + or parsed_mxc_uri.query + or parsed_mxc_uri.params + ): + raise ValueError( + f"Found invalid structure when parsing MXC Uri: {mxc_uri_str}" + ) + + # We use the parsed 'network location' as the server name + server_name = parsed_mxc_uri.netloc + + # urlparse adds a '/' to the beginning of the path, so let's remove that and use + # it as the media_id + media_id = parsed_mxc_uri.path[1:] + + # The media ID should not contain a '/' + if "/" in media_id: + raise ValueError( + f"Found invalid character in media ID portion of MXC Uri: {mxc_uri_str}" + ) + + return cls(server_name, media_id) + + def __str__(self) -> str: + """Convert an MXCUri object to a str.""" + return f"mxc://{self.server_name}/{self.media_id}" diff --git a/tests/types/__init__.py b/tests/types/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/types/test_mxc_uri.py b/tests/types/test_mxc_uri.py new file mode 100644 index 0000000..353d48f --- /dev/null +++ b/tests/types/test_mxc_uri.py @@ -0,0 +1,97 @@ +# Copyright 2022 The Matrix.org Foundation C.I.C. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from unittest import TestCase + +from matrix_common.types.mxc_uri import MXCUri + + +class MXCUriTestCase(TestCase): + def test_valid_mxc_uris_to_str(self) -> None: + """Tests that a series of valid mxc are converted to a str correctly.""" + # Converting an MXCUri to its str representation + mxc_0 = MXCUri(server_name="example.com", media_id="84n8493hnfsjkbcu") + self.assertEqual(str(mxc_0), "mxc://example.com/84n8493hnfsjkbcu") + + mxc_1 = MXCUri( + server_name="192.168.1.17:8008", media_id="bajkad89h31ausdhoqqasd" + ) + self.assertEqual(str(mxc_1), "mxc://192.168.1.17:8008/bajkad89h31ausdhoqqasd") + + mxc_2 = MXCUri(server_name="123.123.123.123", media_id="000000000000") + self.assertEqual(str(mxc_2), "mxc://123.123.123.123/000000000000") + + def test_valid_mxc_uris_from_str(self) -> None: + """Tests that a series of valid mxc uris strs are parsed correctly.""" + # Converting a str to its MXCUri representation + mxcuri_0 = MXCUri.from_str("mxc://example.com/g12789g890ajksjk") + self.assertEqual(mxcuri_0.server_name, "example.com") + self.assertEqual(mxcuri_0.media_id, "g12789g890ajksjk") + + mxcuri_1 = MXCUri.from_str("mxc://localhost:8448/abcdefghijklmnopqrstuvwxyz") + self.assertEqual(mxcuri_1.server_name, "localhost:8448") + self.assertEqual(mxcuri_1.media_id, "abcdefghijklmnopqrstuvwxyz") + + mxcuri_2 = MXCUri.from_str("mxc://[::1]/abcdefghijklmnopqrstuvwxyz") + self.assertEqual(mxcuri_2.server_name, "[::1]") + self.assertEqual(mxcuri_2.media_id, "abcdefghijklmnopqrstuvwxyz") + + mxcuri_3 = MXCUri.from_str("mxc://123.123.123.123:32112/12893y81283781023") + self.assertEqual(mxcuri_3.server_name, "123.123.123.123:32112") + self.assertEqual(mxcuri_3.media_id, "12893y81283781023") + + mxcuri_4 = MXCUri.from_str("mxc://domain/abcdefg") + self.assertEqual(mxcuri_4.server_name, "domain") + self.assertEqual(mxcuri_4.media_id, "abcdefg") + + def test_invalid_mxc_uris_from_str(self) -> None: + """Tests that a series of invalid mxc uris are appropriately rejected.""" + # Converting invalid MXC URI strs to MXCUri representations + with self.assertRaises(ValueError): + MXCUri.from_str("http://example.com/abcdef") + + with self.assertRaises(ValueError): + MXCUri.from_str("mxc:///example.com/abcdef") + + with self.assertRaises(ValueError): + MXCUri.from_str("mxc://example.com//abcdef") + + with self.assertRaises(ValueError): + MXCUri.from_str("mxc://example.com/abcdef/") + + with self.assertRaises(ValueError): + MXCUri.from_str("mxc://example.com/abc/abcdef") + + with self.assertRaises(ValueError): + MXCUri.from_str("mxc://example.com/abc/abcdef") + + with self.assertRaises(ValueError): + MXCUri.from_str("mxc:///abcdef") + + with self.assertRaises(ValueError): + MXCUri.from_str("mxc://example.com") + + with self.assertRaises(ValueError): + MXCUri.from_str("mxc://example.com/") + + with self.assertRaises(ValueError): + MXCUri.from_str("mxc:///") + + with self.assertRaises(ValueError): + MXCUri.from_str("example.com/abc") + + with self.assertRaises(ValueError): + MXCUri.from_str("") + + with self.assertRaises(ValueError): + MXCUri.from_str(None) # type: ignore