diff --git a/plinth/modules/coturn/components.py b/plinth/modules/coturn/components.py index 1e8a83fba..ae05267bf 100644 --- a/plinth/modules/coturn/components.py +++ b/plinth/modules/coturn/components.py @@ -17,7 +17,8 @@ TURN_REST_TTL = 24 * 3600 TURN_REST_USER = 'fbxturnuser' -TURN_URI_REGEX = r'(stun|turn):(.*):([0-9]{4})\?transport=(tcp|udp)' +TURN_URI_REGEX = \ + r'(stun:(.*):([0-9]{4}))|(turn:(.*):([0-9]{4})\?transport=(tcp|udp))' @dataclass @@ -43,8 +44,9 @@ class TurnConfiguration: """Generate URIs after object initialization if necessary.""" if self.domain and not self.uris: self.uris = [ - f'{typ}:{self.domain}:3478?transport={transport}' - for typ in ['stun', 'turn'] for transport in ['tcp', 'udp'] + f'stun:{self.domain}:3478', + f'turn:{self.domain}:3478?transport=tcp', + f'turn:{self.domain}:3478?transport=udp' ] def to_json(self) -> str: @@ -59,7 +61,7 @@ class TurnConfiguration: def validate_turn_uris(turn_uris: list[str]) -> bool: """Return whether the given TURN URI is valid.""" pattern = re.compile(TURN_URI_REGEX) - return all(map(pattern.match, turn_uris)) + return all(map(pattern.fullmatch, turn_uris)) @dataclass diff --git a/plinth/modules/coturn/forms.py b/plinth/modules/coturn/forms.py index e57740c1c..0f6f93608 100644 --- a/plinth/modules/coturn/forms.py +++ b/plinth/modules/coturn/forms.py @@ -18,7 +18,8 @@ def get_domain_choices(): def turn_uris_validator(turn_uris): """Validate list of STUN/TURN Server URIs.""" - if not TurnConfiguration.validate_turn_uris(turn_uris.split("\n")): + uris = [uri for uri in turn_uris.split('\r\n') if uri] + if not TurnConfiguration.validate_turn_uris(uris): raise ValidationError(_('Invalid list of STUN/TURN Server URIs')) diff --git a/plinth/modules/coturn/tests/test_components.py b/plinth/modules/coturn/tests/test_components.py index bf7678888..524cad258 100644 --- a/plinth/modules/coturn/tests/test_components.py +++ b/plinth/modules/coturn/tests/test_components.py @@ -3,6 +3,7 @@ Tests for the Coturn app component. """ +import json from unittest.mock import call, patch import pytest @@ -26,16 +27,15 @@ def fixture_empty_component_list(): TurnConsumer._all = {} -def test_configuration_init(): +def test_turn_configuration_init(): """Test creating configuration object.""" config = TurnConfiguration('test-domain.example', [], 'test-shared-secret') assert config.domain == 'test-domain.example' assert config.shared_secret == 'test-shared-secret' assert config.uris == [ - "stun:test-domain.example:3478?transport=tcp", - "stun:test-domain.example:3478?transport=udp", - "turn:test-domain.example:3478?transport=tcp", - "turn:test-domain.example:3478?transport=udp", + 'stun:test-domain.example:3478', + 'turn:test-domain.example:3478?transport=tcp', + 'turn:test-domain.example:3478?transport=udp', ] config = TurnConfiguration(None, ['test-uri1', 'test-uri2'], @@ -59,6 +59,42 @@ def test_configuration_init(): assert config.credential == 'test-credential' +def test_turn_configuration_to_json(): + """Test exporting configuration to JSON.""" + config = TurnConfiguration('test-domain.example', [], 'test-shared-secret') + assert json.loads(config.to_json()) == { + 'domain': 'test-domain.example', + 'uris': [ + 'stun:test-domain.example:3478', + 'turn:test-domain.example:3478?transport=tcp', + 'turn:test-domain.example:3478?transport=udp', + ], + 'shared_secret': 'test-shared-secret' + } + + +def test_turn_configuration_validate_turn_uris(): + """Test validation method to check for STUN/TURN URIs.""" + valid_uris = [ + 'stun:test-domain.example:3478', + 'turn:test-domain.example:3478?transport=tcp', + 'turn:test-domain.example:3478?transport=udp', + ] + invalid_uris = [ + 'x-invalid-uri', + 'stun:', + 'stun:domain-port-missing.example', + 'stun:testing.example:1234invalid-append', + 'turn:testing.example:1234', + 'turn:testing.example:1234?invalid-param=value', + 'turn:testing.example:1234?transport=invalid-value', + 'turn:testing.example:1234?transport=tcp-invalid-append', + ] + assert TurnConfiguration().validate_turn_uris(valid_uris) + for uri in invalid_uris: + assert not TurnConfiguration().validate_turn_uris([uri]) + + def test_component_init_and_list(): """Test initializing and listing all the components.""" component1 = TurnConsumer('component1')