Implement SIP002 Shadowsocks URI codec

Signed-off-by: Loren Eteval <loren.eteval@proton.me>
This commit is contained in:
Loren Eteval
2026-08-13 23:55:48 +08:00
parent cf8350ffcf
commit ea333d3f3d
6 changed files with 615 additions and 58 deletions
+32 -54
View File
@@ -22,6 +22,12 @@ from __future__ import annotations
from Furious.Models.Configuration import ConfigFactory
from Furious.Models.Encoding import *
from Furious.Models.Protocol import Protocol
from Furious.Backends.ShadowsocksURI import (
SHADOWSOCKS_PLUGIN_METADATA_KEY,
ShadowsocksURIData,
parseShadowsocksURI,
serializeShadowsocksURI,
)
from typing import Union, Tuple
@@ -1255,59 +1261,16 @@ class ConfigXray(ConfigFactory):
@staticmethod
def URI2ProxyOutboundObjectSS(URI: str) -> Tuple[str, dict]:
"""Parse a Shadowsocks URI into a remark and Xray proxy outbound."""
try:
result = urlparse(URI)
remark = unquote(result.fragment)
except Exception:
# Any non-exit exceptions
return '', {}
def getSSParams():
# Begin SIP002...
"""Return ss params."""
try:
# Try pack with 3 element
userinfo, server = result.netloc.split('@')
# Some old SS share link doesn't add padding
# in base64 encoding. Add padding to userinfo
return [
*PyBase64Encoder.decode(userinfo + '===').decode().split(':', 1),
*_parseHostPort(server),
]
except Exception:
# Any non-exit exceptions
pass
try:
# Try pack with 4 element
userinfo, server = result.netloc.split('@')
return [*userinfo.split(':', 1), *_parseHostPort(server)]
except Exception:
# Any non-exit exceptions
pass
try:
# ss://base64...#fragment
userinfo, server = (
PyBase64Encoder.decode(result.netloc).decode().split('@')
)
return [*userinfo.split(':', 1), *_parseHostPort(server)]
except Exception:
# Any non-exit exceptions
pass
raise ValueError(f'Invalid SS URI format {URI}')
parsed = parseShadowsocksURI(URI)
return (
remark,
ConfigXrayProxyOutboundObjectSS(*getSSParams()),
parsed.tag,
ConfigXrayProxyOutboundObjectSS(
parsed.method,
parsed.password,
parsed.host,
parsed.port,
),
)
@staticmethod
@@ -1408,7 +1371,7 @@ class ConfigXray(ConfigFactory):
return super().toJSONString(indent=indent)
def toURI(self, remark: str = '') -> str:
def toURI(self, remark: str = '', **kwargs) -> str:
"""Export the configuration as a share URI."""
override = remark
@@ -1479,10 +1442,25 @@ class ConfigXray(ConfigFactory):
self.proxyServerObject[value]
for value in ['method', 'password', 'address', 'port']
)
plugin = str(kwargs.get('shadowsocksPlugin', '') or '')
profileMetadata = kwargs.get('profileMetadata')
netloc = f'{quote(method)}:{quote(password)}@{address}:{port}'
if not plugin and profileMetadata is not None:
extras = getattr(profileMetadata, 'extras', {})
return urlunparse(['ss', netloc, '', '', '', quote(override)])
if isinstance(extras, dict):
plugin = str(extras.get(SHADOWSOCKS_PLUGIN_METADATA_KEY, '') or '')
return serializeShadowsocksURI(
ShadowsocksURIData(
str(method),
str(password),
str(address),
int(port),
plugin,
override,
)
)
if protocol == 'socks':
address, port = list(
+542
View File
@@ -0,0 +1,542 @@
# Copyright (C) 2024–present Loren Eteval & contributors <loren.eteval@proton.me>
#
# This file is part of Furious.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Parse and serialize Shadowsocks share links according to SIP002."""
from __future__ import annotations
from dataclasses import dataclass
from urllib.parse import quote, unquote_to_bytes, urlsplit
import re
import base64
import binascii
import ipaddress
__all__ = [
'SHADOWSOCKS_PLUGIN_METADATA_KEY',
'ShadowsocksURIData',
'ShadowsocksURIError',
'parseShadowsocksURI',
'serializeShadowsocksURI',
]
SHADOWSOCKS_PLUGIN_METADATA_KEY = 'shadowsocksPlugin'
# Methods implemented by common Shadowsocks clients, including the legacy
# stream ciphers used by existing share links. Keeping validation here
# prevents malformed method names from becoming runtime configurations.
SUPPORTED_SHADOWSOCKS_METHODS = frozenset(
{
'none',
'plain',
'table',
'rc4',
'rc4-md5',
'aes-128-cfb',
'aes-192-cfb',
'aes-256-cfb',
'aes-128-ctr',
'aes-192-ctr',
'aes-256-ctr',
'bf-cfb',
'camellia-128-cfb',
'camellia-192-cfb',
'camellia-256-cfb',
'cast5-cfb',
'des-cfb',
'idea-cfb',
'rc2-cfb',
'salsa20',
'chacha20',
'chacha20-ietf',
'aes-128-gcm',
'aes-192-gcm',
'aes-256-gcm',
'chacha20-poly1305',
'chacha20-ietf-poly1305',
'xchacha20-poly1305',
'xchacha20-ietf-poly1305',
'2022-blake3-aes-128-gcm',
'2022-blake3-aes-256-gcm',
'2022-blake3-chacha20-poly1305',
}
)
_BASE64URL_PATTERN = re.compile(r'^[A-Za-z0-9_-]+={0,2}$')
_HEX_DIGITS = frozenset('0123456789abcdefABCDEF')
_PLUGIN_ESCAPED_CHARACTERS = frozenset('\\:;=')
class ShadowsocksURIError(ValueError):
"""Report a malformed or unsupported Shadowsocks share link."""
@dataclass(frozen=True)
class ShadowsocksURIData:
"""Store the meaningful fields of one SIP002 URI."""
method: str
password: str
host: str
port: int
plugin: str = ''
tag: str = ''
def _validatePercentEncoding(value: str, label: str):
"""Reject incomplete percent escapes before decoding *value*."""
index = 0
while index < len(value):
if value[index] != '%':
index += 1
continue
if (
index + 2 >= len(value)
or value[index + 1] not in _HEX_DIGITS
or value[index + 2] not in _HEX_DIGITS
):
raise ShadowsocksURIError(f'{label} contains invalid percent encoding')
index += 3
def _percentDecode(value: str, label: str) -> str:
"""Decode one RFC3986 component as strict UTF-8 without ``+`` semantics."""
_validatePercentEncoding(value, label)
try:
return unquote_to_bytes(value).decode('utf-8')
except UnicodeDecodeError as ex:
raise ShadowsocksURIError(f'{label} is not valid UTF-8') from ex
def _validateMethodAndPassword(method: str, password: str):
"""Validate fields before constructing a configuration."""
if method not in SUPPORTED_SHADOWSOCKS_METHODS:
raise ShadowsocksURIError(f'unsupported Shadowsocks method {method!r}')
if not password:
raise ShadowsocksURIError('Shadowsocks password cannot be empty')
def _decodeBase64URL(value: str) -> str:
"""Decode padded or unpadded Base64URL using strict alphabet validation."""
if not value or not _BASE64URL_PATTERN.fullmatch(value):
raise ShadowsocksURIError('userinfo is not valid Base64URL')
unpadded = value.rstrip('=')
suppliedPadding = len(value) - len(unpadded)
requiredPadding = -len(unpadded) % 4
if (
'=' in unpadded
or len(unpadded) % 4 == 1
or (suppliedPadding and suppliedPadding != requiredPadding)
):
raise ShadowsocksURIError('userinfo has invalid Base64URL padding')
padded = unpadded + '=' * requiredPadding
try:
decoded = base64.b64decode(
padded.encode('ascii'), altchars=b'-_', validate=True
)
return decoded.decode('utf-8')
except (binascii.Error, UnicodeDecodeError) as ex:
raise ShadowsocksURIError('userinfo is not valid Base64URL UTF-8') from ex
def _decodeLegacyBase64(value: str) -> str:
"""Decode a pre-SIP002 whole-authority payload for import compatibility."""
value = _percentDecode(value, 'legacy payload').rstrip('=')
if not value or len(value) % 4 == 1:
raise ShadowsocksURIError('legacy payload has invalid Base64 padding')
padded = value + '=' * (-len(value) % 4)
try:
decoded = base64.b64decode(
padded.encode('ascii'), altchars=b'-_', validate=True
)
return decoded.decode('utf-8')
except (ValueError, binascii.Error, UnicodeDecodeError) as ex:
raise ShadowsocksURIError('legacy payload is not valid Base64 UTF-8') from ex
def _parseEndpoint(value: str) -> tuple[str, int]:
"""Parse a normal URI host/port authority, including bracketed IPv6."""
try:
result = urlsplit(f'//{value}')
host = result.hostname or ''
port = result.port
except ValueError as ex:
raise ShadowsocksURIError('server endpoint has an invalid port') from ex
if not host or port is None:
raise ShadowsocksURIError('server endpoint must contain a host and port')
if not 1 <= port <= 65535:
raise ShadowsocksURIError('server port must be between 1 and 65535')
if result.username is not None or result.password is not None or result.path:
raise ShadowsocksURIError('server endpoint is malformed')
if any(character.isspace() or ord(character) < 32 for character in host):
raise ShadowsocksURIError('server host contains invalid characters')
return host, port
def _parsePlainUserinfo(value: str) -> tuple[str, str]:
"""Decode SIP002 plain method/password fields."""
if ':' not in value:
raise ShadowsocksURIError('plain userinfo is missing the method separator')
encodedMethod, encodedPassword = value.split(':', 1)
method = _percentDecode(encodedMethod, 'method')
password = _percentDecode(encodedPassword, 'password')
_validateMethodAndPassword(method, password)
return method, password
def _parseUserinfo(value: str) -> tuple[str, str]:
"""Parse SIP002 userinfo and one widespread encoded-separator variant."""
if not value or '@' in value:
raise ShadowsocksURIError('userinfo is malformed')
if ':' in value:
return _parsePlainUserinfo(value)
try:
decoded = _decodeBase64URL(value)
except ShadowsocksURIError as base64Error:
# Some producers percent-encode the entire ``method:password`` value,
# including the separator. It is not canonical SIP002, but accepting
# it preserves established import compatibility; export normalizes it.
decoded = _percentDecode(value, 'userinfo')
if ':' not in decoded:
raise base64Error
method, password = decoded.split(':', 1)
_validateMethodAndPassword(method, password)
if method.startswith('2022-'):
raise ShadowsocksURIError(
'AEAD-2022 userinfo must use a literal method separator'
)
return method, password
if ':' not in decoded:
raise ShadowsocksURIError('Base64URL userinfo is missing the method separator')
method, password = decoded.split(':', 1)
_validateMethodAndPassword(method, password)
if method.startswith('2022-'):
raise ShadowsocksURIError('AEAD-2022 userinfo must not be Base64URL encoded')
return method, password
def _splitEscaped(value: str, separator: str) -> list[str]:
"""Split a SIP003 plugin argument on an unescaped separator."""
result = []
current = []
escaped = False
for character in value:
if escaped:
if character not in _PLUGIN_ESCAPED_CHARACTERS:
raise ShadowsocksURIError(
f'plugin argument contains invalid escape \\{character}'
)
if character == separator:
current.append(character)
else:
# Preserve escapes that belong to a later parsing layer. In
# particular, an escaped equals sign must not become the
# option's key/value delimiter after splitting semicolons.
current.extend(('\\', character))
escaped = False
elif character == '\\':
escaped = True
elif character == separator:
result.append(''.join(current))
current = []
else:
current.append(character)
if escaped:
raise ShadowsocksURIError('plugin argument ends with an incomplete escape')
result.append(''.join(current))
return result
def _splitPluginOption(value: str) -> tuple[str, str | None]:
"""Split and unescape one plugin option at its first unescaped equals."""
parts = _splitEscaped(value, '=')
if len(parts) > 2:
parts[1:] = ['='.join(parts[1:])]
return (
_unescapePluginComponent(parts[0]),
_unescapePluginComponent(parts[1]) if len(parts) == 2 else None,
)
def _unescapePluginComponent(value: str) -> str:
"""Remove validated SIP003 escapes from one component."""
result = []
escaped = False
for character in value:
if escaped:
if character not in _PLUGIN_ESCAPED_CHARACTERS:
raise ShadowsocksURIError(
f'plugin argument contains invalid escape \\{character}'
)
result.append(character)
escaped = False
elif character == '\\':
escaped = True
else:
result.append(character)
if escaped:
raise ShadowsocksURIError('plugin argument ends with an incomplete escape')
return ''.join(result)
def escapePluginComponent(value: str) -> str:
"""Escape one SIP003 plugin name, option name, or option value."""
return ''.join(
f'\\{character}' if character in _PLUGIN_ESCAPED_CHARACTERS else character
for character in str(value)
)
def parsePluginArgument(value: str) -> tuple[str, tuple[tuple[str, str | None], ...]]:
"""Parse one decoded SIP002 plugin argument into semantic components."""
fields = _splitEscaped(value, ';')
name = _unescapePluginComponent(fields[0])
if not name:
raise ShadowsocksURIError('plugin name cannot be empty')
options = []
for field in fields[1:]:
if not field:
raise ShadowsocksURIError('plugin option cannot be empty')
key, optionValue = _splitPluginOption(field)
if not key:
raise ShadowsocksURIError('plugin option name cannot be empty')
options.append((key, optionValue))
return name, tuple(options)
def formatPluginArgument(
name: str, options: tuple[tuple[str, str | None], ...] = tuple()
) -> str:
"""Serialize semantic plugin components with SIP003 escaping."""
if not name:
raise ShadowsocksURIError('plugin name cannot be empty')
fields = [escapePluginComponent(name)]
for key, value in options:
if not key:
raise ShadowsocksURIError('plugin option name cannot be empty')
field = escapePluginComponent(key)
if value is not None:
field += f'={escapePluginComponent(value)}'
fields.append(field)
return ';'.join(fields)
def _normalizePluginArgument(value: str) -> str:
"""Validate and canonically re-escape one decoded plugin argument."""
name, options = parsePluginArgument(value)
return formatPluginArgument(name, options)
def _parseQuery(value: str) -> str:
"""Return the first plugin query value and ignore unsupported parameters."""
for field in value.split('&') if value else tuple():
encodedKey, separator, encodedValue = field.partition('=')
try:
key = _percentDecode(encodedKey, 'query parameter name')
except ShadowsocksURIError:
# An unsupported malformed parameter must not invalidate a valid
# Shadowsocks URI.
continue
if key != 'plugin':
continue
if not separator:
raise ShadowsocksURIError('plugin query parameter has no value')
return _normalizePluginArgument(
_percentDecode(encodedValue, 'plugin query parameter')
)
return ''
def _parseLegacyURI(result) -> ShadowsocksURIData:
"""Parse the pre-SIP002 ``base64(method:password@host:port)`` form."""
# A standard Base64 payload may itself contain ``/``. ``urlsplit`` moves
# that suffix into ``path``, so joining the two components reconstructs the
# original payload without treating any Base64 character as URI structure.
payload = f'{result.netloc}{result.path}'
decoded = _decodeLegacyBase64(payload)
userinfo, separator, endpoint = decoded.rpartition('@')
if not separator:
raise ShadowsocksURIError('legacy payload has no server separator')
method, password = _parsePlainUserinfo(userinfo)
host, port = _parseEndpoint(endpoint)
tag = _percentDecode(result.fragment, 'tag')
return ShadowsocksURIData(method, password, host, port, tag=tag)
def parseShadowsocksURI(uri: str) -> ShadowsocksURIData:
"""Parse a SIP002 URI, with isolated pre-SIP002 import compatibility."""
if not isinstance(uri, str) or not uri.strip():
raise ShadowsocksURIError('Shadowsocks URI cannot be empty')
uri = uri.strip()
try:
result = urlsplit(uri)
except ValueError as ex:
raise ShadowsocksURIError('Shadowsocks URI is malformed') from ex
if result.scheme.casefold() != 'ss':
raise ShadowsocksURIError('URI scheme must be ss')
if '@' not in result.netloc:
return _parseLegacyURI(result)
if result.path not in ('', '/'):
raise ShadowsocksURIError('SIP002 URI path must be empty or /')
userinfo, separator, endpoint = result.netloc.rpartition('@')
if not separator:
raise ShadowsocksURIError('SIP002 URI is missing userinfo')
method, password = _parseUserinfo(userinfo)
host, port = _parseEndpoint(endpoint)
plugin = _parseQuery(result.query)
tag = _percentDecode(result.fragment, 'tag')
return ShadowsocksURIData(method, password, host, port, plugin, tag)
def _formatHost(host: str) -> str:
"""Return a canonical URI host, bracketing IPv6 literals."""
host = str(host).strip()
if host.startswith('[') and host.endswith(']'):
host = host[1:-1]
if not host:
raise ShadowsocksURIError('server host cannot be empty')
try:
address = ipaddress.ip_address(host)
except ValueError:
if ':' in host or any(
character.isspace() or ord(character) < 32 for character in host
):
raise ShadowsocksURIError('server host is malformed')
try:
return host.encode('idna').decode('ascii')
except UnicodeError as ex:
raise ShadowsocksURIError('server host is malformed') from ex
return f'[{address.compressed}]' if address.version == 6 else address.compressed
def serializeShadowsocksURI(value: ShadowsocksURIData) -> str:
"""Serialize one configuration into canonical SIP002 form."""
if not isinstance(value, ShadowsocksURIData):
raise TypeError('value must be a ShadowsocksURIData instance')
_validateMethodAndPassword(value.method, value.password)
try:
port = int(value.port)
except (TypeError, ValueError) as ex:
raise ShadowsocksURIError('server port must be an integer') from ex
if not 1 <= port <= 65535:
raise ShadowsocksURIError('server port must be between 1 and 65535')
if value.method.startswith('2022-'):
userinfo = f'{quote(value.method, safe="")}:{quote(value.password, safe="")}'
else:
encoded = base64.urlsafe_b64encode(
f'{value.method}:{value.password}'.encode('utf-8')
).decode('ascii')
userinfo = encoded.rstrip('=')
endpoint = f'{_formatHost(value.host)}:{port}'
path, query = '', ''
if value.plugin:
plugin = _normalizePluginArgument(value.plugin)
path, query = '/', f'plugin={quote(plugin, safe="")}'
fragment = quote(value.tag, safe='')
return f'ss://{userinfo}@{endpoint}{path}{"?" + query if query else ""}{"#" + fragment if fragment else ""}'
+25 -1
View File
@@ -24,6 +24,10 @@ from Furious.Backends.Configuration import (
ConfigXray,
configXrayEmptyProxyOutboundObject,
)
from Furious.Backends.ShadowsocksURI import (
SHADOWSOCKS_PLUGIN_METADATA_KEY,
parseShadowsocksURI,
)
from Furious.Plugins.API import (
ProtocolDescriptor,
ProtocolHandler,
@@ -65,6 +69,13 @@ class XrayProtocolHandler(ProtocolHandler):
"""Parse this handler's URI directly into an Xray configuration."""
parser = getattr(ConfigXray, self._parserName)
remark, proxyOutbound = parser(uri)
metadata = {'displayName': remark}
if self.protocolId == 'shadowsocks':
plugin = parseShadowsocksURI(uri).plugin
if plugin:
metadata[SHADOWSOCKS_PLUGIN_METADATA_KEY] = plugin
if (
not proxyOutbound
@@ -77,7 +88,7 @@ class XrayProtocolHandler(ProtocolHandler):
return ProtocolParseResult(
ConfigXray(config),
{'displayName': remark},
metadata,
)
def fromMapping(self, configuration, **kwargs):
@@ -108,6 +119,19 @@ class XrayProtocolHandler(ProtocolHandler):
"""Export an owned Xray configuration to its share-link format."""
return configuration.toURI(remark) if self.supports(configuration) else ''
def exportProfile(self, profile, remark: str = '') -> str:
"""Export Xray profile metadata required by protocol share formats."""
configuration = getattr(profile, 'connection', profile)
return (
configuration.toURI(
remark,
profileMetadata=getattr(profile, 'metadata', None),
)
if self.supports(configuration)
else ''
)
def _placeholder(x):
return x
+9 -1
View File
@@ -251,7 +251,15 @@ class ServerProfile(MutableMapping[str, Any]):
def toURI(self, remark: str = '') -> str:
"""Serialize the connection document as a share URI."""
return self.connection.toURI(remark or self.metadata.displayName)
exportOptions = (
{'profileMetadata': self.metadata}
if self.itemProtocol.casefold() == 'shadowsocks'
else {}
)
return self.connection.toURI(
remark or self.metadata.displayName, **exportOptions
)
def httpProxy(self) -> str:
"""Return the connection's HTTP proxy endpoint."""
+4
View File
@@ -295,6 +295,10 @@ class ProtocolHandler(PluginCapability):
"""Serialize one owned configuration to a share URI."""
return ''
def exportProfile(self, profile, remark: str = '') -> str:
"""Serialize a profile while keeping older handlers source-compatible."""
return self.export(getattr(profile, 'connection', profile), remark)
def validate(self, configuration) -> Tuple[str, ...]:
"""Return validation errors for one owned configuration."""
if not self.supports(configuration):
+3 -2
View File
@@ -670,7 +670,8 @@ class PluginRegistry:
# Any non-exit exceptions
logger.error(
f'failed to parse {handler.descriptor.id!r} configuration: {ex}'
f'failed to parse {handler.descriptor.id!r} configuration: {ex}. '
f'URI: {uri!r}'
)
return None
@@ -828,7 +829,7 @@ class PluginRegistry:
if not remark:
remark = str(getattr(config, 'itemRemark', ''))
return handler.export(_connectionOf(config), remark)
return handler.exportProfile(config, remark)
def validateConfig(self, config):
"""Validate a configuration through its protocol capability."""