feat(http): expose optional tunnel CORS policies

README.md

@@ -310,3 +310,9 @@ Tests in `tests/integration/test_engine_mtls.py` use real TLS servers for discov
 inventory, both watch transports, trust validation and cancellation. Opt-in
 real-Engine inventory checks use `RSTREAM_PYTHON_E2E_ENGINE_API=1` with the
 existing E2E credential configuration.
+
+## HTTP CORS
+
+Pass `http_cors=HTTPCORS(allowed_origins=("https://dashboard.example",), allowed_methods=("GET", "PUT"), allowed_headers=("Authorization",), exposed_headers=("ETag",), max_age=600)` to `create_tunnel` or `CreateTunnelOptions`. Import `HTTPCORS` from `rstream`.
+
+CORS requires a published HTTP tunnel. Allowed origins must be explicit HTTP(S) origins without paths or wildcards. Omitted method and request-header lists use GET/HEAD/POST and Authorization/Content-Type; credentials are disabled and max age is zero by default. An absent policy preserves upstream CORS. Engine answers valid preflights before request authentication, but every actual request still requires its usual authorization. No filesystem or upstream access is granted by a preflight. Deploy an Engine supporting protocol patch 1.4.6 before using this SDK version; the new Engine remains compatible with supported older clients.

proto/rstream.proto

@@ -14,7 +14,7 @@ extend google.protobuf.FieldOptions {
   string access = 51234;
 }
 
-option (protocol_version) = "1.4.5";
+option (protocol_version) = "1.4.6";
 
 package rstream.io_rstrm.protobuf;
 
@@ -64,6 +64,7 @@ enum ErrorCode {
 
   // 7000-7999: Runtime capacity exhausted
   ERROR_CODE_CAPACITY_EXHAUSTED = 7000;
+  ERROR_CODE_RESOURCE_CONFLICT = 8000;
 
   // 9000-9999: Internal errors
   ERROR_CODE_INTERNAL = 9000;
@@ -102,6 +103,16 @@ message ServerDetails {
 
 // Tunnel message structure used for creating and managing tunnels
 
+// Optional browser access policy for a published HTTP tunnel.
+message HttpCors {
+  repeated string allowed_origins = 1;
+  repeated string allowed_methods = 2;
+  repeated string allowed_headers = 3;
+  repeated string exposed_headers = 4;
+  bool allow_credentials = 5;
+  int64 max_age = 6;
+}
+
 message TunnelProperties {
   google.protobuf.StringValue id = 1 [(access) = "read-only"];
   google.protobuf.Timestamp creation_date = 2 [(access) = "read-only"];
@@ -129,6 +140,7 @@ message TunnelProperties {
   google.protobuf.BoolValue upstream_tls = 24 [(access) = "read-write"];
   google.protobuf.BoolValue datagram_guaranteed_delivery = 25 [(access) = "read-write"];
   google.protobuf.BoolValue allow_cross_region_routing = 26 [(access) = "read-write"];
+  HttpCors http_cors = 27 [(access) = "read-write"];
 }
 
 // When a client opens a new control channel to the server

src/rstream/__init__.py

@@ -27,6 +27,7 @@
 from rstream.stream import RstreamStream
 from rstream.tunnel import BytestreamTunnel, format_forwarding_address
 from rstream.types import (
+    HTTPCORS,
     CreateTunnelOptions,
     ServerDetails,
     TunnelAuth,
@@ -52,6 +53,7 @@ def __getattr__(name: str) -> ModuleType:
 
 
 __all__ = [
+    "HTTPCORS",
     "BytestreamTunnel",
     "Client",
     "ClientFilters",

src/rstream/_proto/rstream_pb2.pyi

@@ -25,6 +25,7 @@ class ErrorCode(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
     ERROR_CODE_FEATURE_NOT_AVAILABLE: _ClassVar[ErrorCode]
     ERROR_CODE_SERVICE_UNAVAILABLE: _ClassVar[ErrorCode]
     ERROR_CODE_CAPACITY_EXHAUSTED: _ClassVar[ErrorCode]
+    ERROR_CODE_RESOURCE_CONFLICT: _ClassVar[ErrorCode]
     ERROR_CODE_INTERNAL: _ClassVar[ErrorCode]
 ERROR_CODE_UNSPECIFIED: ErrorCode
 ERROR_CODE_UNAUTHORIZED: ErrorCode
@@ -37,6 +38,7 @@ ERROR_CODE_INVALID_STREAM: ErrorCode
 ERROR_CODE_FEATURE_NOT_AVAILABLE: ErrorCode
 ERROR_CODE_SERVICE_UNAVAILABLE: ErrorCode
 ERROR_CODE_CAPACITY_EXHAUSTED: ErrorCode
+ERROR_CODE_RESOURCE_CONFLICT: ErrorCode
 ERROR_CODE_INTERNAL: ErrorCode
 PROTOCOL_VERSION_FIELD_NUMBER: _ClassVar[int]
 protocol_version: _descriptor.FieldDescriptor
@@ -95,8 +97,24 @@ class ServerDetails(_message.Message):
     update: _wrappers_pb2.StringValue
     def __init__(self, agent: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., channel: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., version: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., plan: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., provider: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., region: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., update: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ...) -> None: ...
 
+class HttpCors(_message.Message):
+    __slots__ = ("allowed_origins", "allowed_methods", "allowed_headers", "exposed_headers", "allow_credentials", "max_age")
+    ALLOWED_ORIGINS_FIELD_NUMBER: _ClassVar[int]
+    ALLOWED_METHODS_FIELD_NUMBER: _ClassVar[int]
+    ALLOWED_HEADERS_FIELD_NUMBER: _ClassVar[int]
+    EXPOSED_HEADERS_FIELD_NUMBER: _ClassVar[int]
+    ALLOW_CREDENTIALS_FIELD_NUMBER: _ClassVar[int]
+    MAX_AGE_FIELD_NUMBER: _ClassVar[int]
+    allowed_origins: _containers.RepeatedScalarFieldContainer[str]
+    allowed_methods: _containers.RepeatedScalarFieldContainer[str]
+    allowed_headers: _containers.RepeatedScalarFieldContainer[str]
+    exposed_headers: _containers.RepeatedScalarFieldContainer[str]
+    allow_credentials: bool
+    max_age: int
+    def __init__(self, allowed_origins: _Optional[_Iterable[str]] = ..., allowed_methods: _Optional[_Iterable[str]] = ..., allowed_headers: _Optional[_Iterable[str]] = ..., exposed_headers: _Optional[_Iterable[str]] = ..., allow_credentials: _Optional[bool] = ..., max_age: _Optional[int] = ...) -> None: ...
+
 class TunnelProperties(_message.Message):
-    __slots__ = ("id", "creation_date", "name", "type", "publish", "protocol", "labels", "geoip", "trusted_ips", "host", "tls_mode", "tls_alpns", "tls_min_version", "tls_ciphers", "mtls_auth", "mtls_cacert_pem", "http_version", "http_use_tls", "token_auth", "rstream_auth", "challenge_mode", "hostname", "port", "upstream_tls", "datagram_guaranteed_delivery", "allow_cross_region_routing")
+    __slots__ = ("id", "creation_date", "name", "type", "publish", "protocol", "labels", "geoip", "trusted_ips", "host", "tls_mode", "tls_alpns", "tls_min_version", "tls_ciphers", "mtls_auth", "mtls_cacert_pem", "http_version", "http_use_tls", "token_auth", "rstream_auth", "challenge_mode", "hostname", "port", "upstream_tls", "datagram_guaranteed_delivery", "allow_cross_region_routing", "http_cors")
     class LabelsEntry(_message.Message):
         __slots__ = ("key", "value")
         KEY_FIELD_NUMBER: _ClassVar[int]
@@ -130,6 +148,7 @@ class TunnelProperties(_message.Message):
     UPSTREAM_TLS_FIELD_NUMBER: _ClassVar[int]
     DATAGRAM_GUARANTEED_DELIVERY_FIELD_NUMBER: _ClassVar[int]
     ALLOW_CROSS_REGION_ROUTING_FIELD_NUMBER: _ClassVar[int]
+    HTTP_CORS_FIELD_NUMBER: _ClassVar[int]
     id: _wrappers_pb2.StringValue
     creation_date: _timestamp_pb2.Timestamp
     name: _wrappers_pb2.StringValue
@@ -156,7 +175,8 @@ class TunnelProperties(_message.Message):
     upstream_tls: _wrappers_pb2.BoolValue
     datagram_guaranteed_delivery: _wrappers_pb2.BoolValue
     allow_cross_region_routing: _wrappers_pb2.BoolValue
-    def __init__(self, id: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., creation_date: _Optional[_Union[datetime.datetime, _timestamp_pb2.Timestamp, _Mapping]] = ..., name: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., type: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., publish: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., protocol: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., labels: _Optional[_Mapping[str, str]] = ..., geoip: _Optional[_Iterable[str]] = ..., trusted_ips: _Optional[_Iterable[str]] = ..., host: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., tls_mode: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., tls_alpns: _Optional[_Iterable[str]] = ..., tls_min_version: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., tls_ciphers: _Optional[_Iterable[str]] = ..., mtls_auth: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., mtls_cacert_pem: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., http_version: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., http_use_tls: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., token_auth: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., rstream_auth: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., challenge_mode: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., hostname: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., port: _Optional[_Union[_wrappers_pb2.UInt32Value, _Mapping]] = ..., upstream_tls: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., datagram_guaranteed_delivery: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., allow_cross_region_routing: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ...) -> None: ...
+    http_cors: HttpCors
+    def __init__(self, id: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., creation_date: _Optional[_Union[datetime.datetime, _timestamp_pb2.Timestamp, _Mapping]] = ..., name: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., type: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., publish: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., protocol: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., labels: _Optional[_Mapping[str, str]] = ..., geoip: _Optional[_Iterable[str]] = ..., trusted_ips: _Optional[_Iterable[str]] = ..., host: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., tls_mode: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., tls_alpns: _Optional[_Iterable[str]] = ..., tls_min_version: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., tls_ciphers: _Optional[_Iterable[str]] = ..., mtls_auth: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., mtls_cacert_pem: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., http_version: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., http_use_tls: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., token_auth: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., rstream_auth: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., challenge_mode: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., hostname: _Optional[_Union[_wrappers_pb2.StringValue, _Mapping]] = ..., port: _Optional[_Union[_wrappers_pb2.UInt32Value, _Mapping]] = ..., upstream_tls: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., datagram_guaranteed_delivery: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., allow_cross_region_routing: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] = ..., http_cors: _Optional[_Union[HttpCors, _Mapping]] = ...) -> None: ...
 
 class OpenControlChannelReq(_message.Message):
     __slots__ = ("client_details", "liveness")

src/rstream/control.py

@@ -26,6 +26,7 @@
 from rstream.stream import RstreamStream
 from rstream.tunnel import BytestreamTunnel
 from rstream.types import (
+    HTTPCORS,
     CreateTunnelOptions,
     HTTPVersion,
     ServerDetails,
@@ -125,6 +126,7 @@ async def create_tunnel(
         upstream_tls: bool | None = None,
         allow_cross_region_routing: bool | None = None,
         auth: TunnelAuth | None = None,
+        http_cors: HTTPCORS | None = None,
     ) -> BytestreamTunnel:
         explicit_options = CreateTunnelOptions(
             name=name,
@@ -148,6 +150,7 @@ async def create_tunnel(
             upstream_tls=upstream_tls,
             allow_cross_region_routing=allow_cross_region_routing,
             auth=auth,
+            http_cors=http_cors,
         )
         if options is not None and explicit_options != CreateTunnelOptions():
             raise RuntimeError(
@@ -494,6 +497,13 @@ def _preserve_payloads_after(error: BaseException) -> bool:
 
 
 def _normalize_bytestream_options(options: CreateTunnelOptions) -> TunnelProperties:
+    if options.http_cors is not None and (
+        options.publish is False or options.protocol not in (None, "http")
+    ):
+        raise RuntimeError(
+            "CORS requires a published HTTP tunnel.",
+            code="ERR_RSTREAM_INVALID_TUNNEL",
+        )
     if options.port is not None and options.protocol != "tcp":
         raise RuntimeError(
             "A published port requires protocol='tcp'.",
@@ -564,6 +574,7 @@ def _normalize_bytestream_options(options: CreateTunnelOptions) -> TunnelPropert
         tls_ciphers=tuple(options.tls_ciphers),
         mtls_auth=options.mtls_auth,
         http_version=options.http_version,
+        http_cors=options.http_cors,
         token_auth=token_auth,
         rstream_auth=rstream_auth,
         challenge_mode=challenge_mode,

src/rstream/engine_api.py

@@ -16,7 +16,7 @@
 
 from rstream.config import TLSOptions, create_ssl_context
 from rstream.errors import RuntimeError
-from rstream.types import TunnelProperties
+from rstream.types import HTTPCORS, TunnelProperties
 
 WatchTransport = Literal["sse", "websocket"]
 
@@ -149,7 +149,9 @@ def tunnel_inventory_from_json(data: Mapping[str, object]) -> TunnelInventory:
 
 def tunnel_properties_from_json(data: Mapping[str, object]) -> TunnelProperties:
     labels = data.get("labels")
+    cors = data.get("http_cors")
     return TunnelProperties(
+        http_cors=_http_cors_from_json(cors) if isinstance(cors, Mapping) else None,
         id=_optional_str(data, "id"),
         creation_date=_optional_datetime(data, "creation_date"),
         name=_optional_str(data, "name"),
@@ -431,3 +433,14 @@ def _string_tuple(data: Mapping[str, object], key: str) -> tuple[str, ...]:
     if not isinstance(value, list):
         return ()
     return tuple(item for item in value if isinstance(item, str))
+
+
+def _http_cors_from_json(data: Mapping[str, object]) -> HTTPCORS:
+    return HTTPCORS(
+        allowed_origins=_string_tuple(data, "allowed_origins"),
+        allowed_methods=_string_tuple(data, "allowed_methods"),
+        allowed_headers=_string_tuple(data, "allowed_headers"),
+        exposed_headers=_string_tuple(data, "exposed_headers"),
+        allow_credentials=_optional_bool(data, "allow_credentials") or False,
+        max_age=_optional_int(data, "max_age") or 0,
+    )

src/rstream/protocol.py

@@ -15,6 +15,7 @@
 from rstream._proto import rstream_pb2 as pb
 from rstream.errors import EngineError, ProtocolError
 from rstream.types import (
+    HTTPCORS,
     HTTPVersion,
     ServerDetails,
     TLSMode,
@@ -159,11 +160,34 @@ def tunnel_properties_to_pb(properties: TunnelProperties) -> pb.TunnelProperties
         result.allow_cross_region_routing.CopyFrom(
             BoolValue(value=properties.allow_cross_region_routing)
         )
+    if properties.http_cors is not None:
+        config = properties.http_cors
+        result.http_cors.CopyFrom(
+            pb.HttpCors(
+                allowed_origins=config.allowed_origins,
+                allowed_methods=config.allowed_methods,
+                allowed_headers=config.allowed_headers,
+                exposed_headers=config.exposed_headers,
+                allow_credentials=config.allow_credentials,
+                max_age=config.max_age,
+            )
+        )
     return result
 
 
 def tunnel_properties_from_pb(properties: pb.TunnelProperties) -> TunnelProperties:
+    config = properties.http_cors
     return TunnelProperties(
+        http_cors=HTTPCORS(
+            allowed_origins=tuple(config.allowed_origins),
+            allowed_methods=tuple(config.allowed_methods),
+            allowed_headers=tuple(config.allowed_headers),
+            exposed_headers=tuple(config.exposed_headers),
+            allow_credentials=config.allow_credentials,
+            max_age=config.max_age,
+        )
+        if properties.HasField("http_cors")
+        else None,
         id=wrapper_string(optional_field(properties, "id", StringValue)),
         creation_date=wrapper_datetime(
             optional_field(properties, "creation_date", Timestamp)

src/rstream/types.py

@@ -22,6 +22,18 @@ class TunnelAuth:
     challenge: bool | None = None
 
 
+@dataclass(frozen=True)
+class HTTPCORS:
+    """Optional browser access policy for a published HTTP tunnel."""
+
+    allowed_origins: tuple[str, ...]
+    allowed_methods: tuple[str, ...] = ()
+    allowed_headers: tuple[str, ...] = ()
+    exposed_headers: tuple[str, ...] = ()
+    allow_credentials: bool = False
+    max_age: int = 0
+
+
 @dataclass(frozen=True)
 class TunnelProperties:
     """Tunnel properties accepted by the engine."""
@@ -51,6 +63,7 @@ class TunnelProperties:
     upstream_tls: bool | None = None
     datagram_guaranteed_delivery: bool | None = None
     allow_cross_region_routing: bool | None = None
+    http_cors: HTTPCORS | None = None
 
 
 @dataclass(frozen=True)
@@ -78,6 +91,7 @@ class CreateTunnelOptions:
     upstream_tls: bool | None = None
     allow_cross_region_routing: bool | None = None
     auth: TunnelAuth | None = None
+    http_cors: HTTPCORS | None = None
 
 
 @dataclass(frozen=True)

tests/unit/test_http_cors.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import pytest
+
+from rstream import HTTPCORS, CreateTunnelOptions, TunnelProperties
+from rstream._proto import rstream_pb2 as pb
+from rstream.control import _normalize_bytestream_options
+from rstream.engine_api import tunnel_properties_from_json
+from rstream.errors import RuntimeError
+from rstream.protocol import tunnel_properties_from_pb, tunnel_properties_to_pb
+
+
+def test_policy_round_trip_and_absence() -> None:
+    policy = HTTPCORS(
+        allowed_origins=("https://dashboard.example",),
+        allowed_methods=("GET", "PUT"),
+        allowed_headers=("Authorization",),
+        exposed_headers=("ETag",),
+        allow_credentials=True,
+        max_age=600,
+    )
+    properties = _normalize_bytestream_options(CreateTunnelOptions(http_cors=policy))
+    wire = tunnel_properties_to_pb(properties).SerializeToString()
+    decoded = tunnel_properties_from_pb(pb.TunnelProperties.FromString(wire))
+    assert decoded.http_cors == policy
+    assert not tunnel_properties_to_pb(TunnelProperties()).HasField("http_cors")
+    assert tunnel_properties_from_pb(pb.TunnelProperties()).http_cors is None
+    assert tunnel_properties_from_json(
+        {
+            "http_cors": {
+                "allowed_origins": ["https://dashboard.example"],
+                "max_age": 120,
+            }
+        }
+    ).http_cors == HTTPCORS(("https://dashboard.example",), max_age=120)
+    assert tunnel_properties_from_json({}).http_cors is None
+
+
+@pytest.mark.parametrize(
+    "options",
+    [
+        CreateTunnelOptions(
+            protocol="tls", http_cors=HTTPCORS(("https://dashboard.example",))
+        ),
+        CreateTunnelOptions(
+            publish=False, http_cors=HTTPCORS(("https://dashboard.example",))
+        ),
+    ],
+)
+def test_policy_requires_published_http(options: CreateTunnelOptions) -> None:
+    with pytest.raises(RuntimeError, match="published HTTP"):
+        _normalize_bytestream_options(options)