Skip to content

Commit

Permalink
Improved tests validating v2 tunnel definition configs for HTTP, TCP,…
Browse files Browse the repository at this point in the history 
… and TLS.
  • Loading branch information
@alexdlaird
alexdlaird committed Dec 31, 2024
1 parent 8da924f commit 0755638
Show file tree
Hide file tree
Showing 4 changed files with 226 additions and 36 deletions.
4 changes: 3 additions & 1 deletion pyngrok/conf.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,8 @@ def __init__(self,
request_timeout: float = 4,
start_new_session: bool = False,
ngrok_version: str = "v3",
api_key: Optional[str] = None) -> None:
api_key: Optional[str] = None,
config_version: int = "2") -> None:
#: The path to the ``ngrok`` binary, defaults to being placed in the same directory as
#: `ngrok's configs <https://ngrok.com/docs/agent/config/v2>`_.
self.ngrok_path: str = DEFAULT_NGROK_PATH if ngrok_path is None else ngrok_path
Expand Down Expand Up @@ -88,6 +89,7 @@ def __init__(self,
self.ngrok_version: str = ngrok_version
#: A ``ngrok`` API key.
self.api_key: Optional[str] = api_key
self.config_version = config_version


_default_pyngrok_config: PyngrokConfig = PyngrokConfig()
Expand Down
16 changes: 10 additions & 6 deletions pyngrok/installer.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,8 @@ def _install_ngrok_zip(ngrok_path: str,

def get_ngrok_config(config_path: str,
use_cache: bool = True,
ngrok_version: Optional[str] = "v3") -> Dict[str, Any]:
ngrok_version: Optional[str] = "v3",
config_version: Optional[int] = "2") -> Dict[str, Any]:
"""
Get the ``ngrok`` config from the given path.
Expand All @@ -161,32 +162,35 @@ def get_ngrok_config(config_path: str,
with open(config_path, "r") as config_file:
config = yaml.safe_load(config_file)
if config is None:
config = get_default_config(ngrok_version)
config = get_default_config(ngrok_version, config_version)

_config_cache[config_path] = config

return _config_cache[config_path]


def get_default_config(ngrok_version: Optional[str]) -> Dict[str, Any]:
def get_default_config(ngrok_version: Optional[str],
config_version: Optional[str]) -> Dict[str, Any]:
"""
Get the default config params for the given major version of ``ngrok``.
:param ngrok_version: The major version of ``ngrok`` installed.
:param config_version: The ``ngrok`` config version.
:return: The default config.
:raises: :class:`~pyngrok.exception.PyngrokError`: When the ``ngrok_version`` is not supported.
"""
if ngrok_version == "v2":
return {}
elif ngrok_version == "v3":
return {"version": "2", "region": "us"}
return {"version": config_version}
else:
raise PyngrokError(f"\"ngrok_version\" must be a supported version: {SUPPORTED_NGROK_VERSIONS}")


def install_default_config(config_path: str,
data: Optional[Dict[str, Any]] = None,
ngrok_version: Optional[str] = "v3") -> None:
ngrok_version: Optional[str] = "v3",
config_version: Optional[str] = "2") -> None:
"""
Install the given data to the ``ngrok`` config. If a config is not already present for the given path, create one.
Before saving new data to the default config, validate that they are compatible with ``pyngrok``.
Expand All @@ -200,7 +204,7 @@ def install_default_config(config_path: str,
else:
data = copy.deepcopy(data)

data.update(get_default_config(ngrok_version))
data.update(get_default_config(ngrok_version, config_version))

config_dir = os.path.dirname(config_path)
if not os.path.exists(config_dir):
Expand Down
2 changes: 1 addition & 1 deletion pyngrok/ngrok.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -197,7 +197,7 @@ def _interpolate_tunnel_definition(pyngrok_config: PyngrokConfig,
if os.path.exists(config_path):
config = installer.get_ngrok_config(config_path, ngrok_version=pyngrok_config.ngrok_version)
else:
config = get_default_config(pyngrok_config.ngrok_version)
config = get_default_config(pyngrok_config.ngrok_version, pyngrok_config.config_version)

tunnel_definitions = config.get("tunnels", {})
# If a "pyngrok-default" tunnel definition exists in the ngrok config, use that
Expand Down
240 changes: 212 additions & 28 deletions tests/test_ngrok.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -910,6 +910,212 @@ def test_upgrade_ngrok_config_file_v2_to_v3(self):
# WHEN
ngrok.connect(pyngrok_config=pyngrok_config_v3)

@unittest.skipIf(not os.environ.get("NGROK_AUTHTOKEN"), "NGROK_AUTHTOKEN environment variable not set")
def test_full_v2_http_tunnel_definitions(self):
# GIVEN
config = {
"version": "2",
"tunnels": {
"my-tunnel": {
"addr": "5000",
"metadata": "metadata",
"inspect": False,
"proto": "http",
"basic_auth": ["auth-token"],
"circuit_breaker": 20,
"compression": False,
"host_header": "host-header",
"domain": "pyngrok.com",
"ip_restriction": {"allow_cidrs": ["allowed"], "deny_cidrs": ["denied"]},
"mutual_tls_cas": __file__,
"oauth": {
"provider": "google",
"allow_domains": ["pyngrok.com"],
"allow_emails": ["email@pyngrok.com"]
},
"proxy_proto": "1",
"request_header": {"add": ["req-addition"], "remove": ["req-subtraction"]},
"response_header": {"add": ["res-addition"], "remove": ["res-subtraction"]},
"schemes": ["https"],
# Can't be used in conjunction with "domain", but validated in other tests
# "subdomain": "pyngrok",
"traffic_policy":
{"on_http_request": {"name": "inbound-policy", "expressions": "inbound-policy-expression",
"actions": {"type": "inbound-policy-actions-type",
"config": "inbound-policy-actions-config"}},
"on_http_response": {"name": "outbound-policy", "expressions": "outbound-policy-expression",
"actions": {"type": "outbound-policy-actions-type",
"config": "outbound-policy-actions-config"}}},
"user_agent_filter": {"allow": ["allow-user-agent"], "deny": ["deny-user-agent"]},
"verify_webhook": {"provider": "provider", "secret": "secret"},
"websocket_tcp_converter": False
}
}
}
config_path = os.path.join(self.config_dir, "config_v3_2.yml")
installer.install_default_config(config_path, config, ngrok_version="v3")
pyngrok_config = self.copy_with_updates(self.pyngrok_config_v3,
api_key="api-key",
config_path=config_path)

# WHEN
# This test ensures the validity of the config (pyngrok will fail with PyngrokNgrokError if the above is
# invalid), not that the config can start a valid tunnel (thus PyngrokNgrokHTTPError is thrown when pyngrok
# tries to open the tunnel after ngrok successfully starts, but we can ignore that for this test
with self.assertRaises(PyngrokNgrokHTTPError):
ngrok.connect(name="my-tunnel", pyngrok_config=pyngrok_config)

@unittest.skipIf(not os.environ.get("NGROK_AUTHTOKEN"), "NGROK_AUTHTOKEN environment variable not set")
def test_full_v2_tcp_tunnel_definitions(self):
# GIVEN
config = {
"version": "2",
"tunnels": {
"my-tunnel": {
"addr": "5000",
"metadata": "metadata",
"inspect": False,
"proto": "tcp",
"ip_restriction": {"allow_cidrs": ["allowed"], "deny_cidrs": ["denied"]},
"proxy_proto": "1",
"remote_addr": "2.tcp.ngrok.io:21746",
"traffic_policy":
{"inbound": {"name": "inbound-policy", "expressions": "inbound-policy-expression",
"actions": {"type": "inbound-policy-actions-type",
"config": "inbound-policy-actions-config"}},
"outbound": {"name": "outbound-policy", "expressions": "outbound-policy-expression",
"actions": {"type": "outbound-policy-actions-type",
"config": "outbound-policy-actions-config"}}}
}
}
}
config_path = os.path.join(self.config_dir, "config_v3_2.yml")
installer.install_default_config(config_path, config, ngrok_version="v3")
pyngrok_config = self.copy_with_updates(self.pyngrok_config_v3,
api_key="api-key",
config_path=config_path)

# WHEN
# This test ensures the validity of the config (pyngrok will fail with PyngrokNgrokError if the above is
# invalid), not that the config can start a valid tunnel (thus PyngrokNgrokHTTPError is thrown when pyngrok
# tries to open the tunnel after ngrok successfully starts, but we can ignore that for this test
with self.assertRaises(PyngrokNgrokHTTPError):
ngrok.connect(name="my-tunnel", pyngrok_config=pyngrok_config)

@unittest.skipIf(not os.environ.get("NGROK_AUTHTOKEN"), "NGROK_AUTHTOKEN environment variable not set")
def test_full_v2_tls_tunnel_definitions(self):
# GIVEN
config = {
"version": "2",
"tunnels": {
"my-tunnel": {
"addr": "5000",
"metadata": "metadata",
"inspect": False,
"proto": "tls",
# "mutual_tls_cas": __file__,
# "crt": __file__,
"domain": "pyngrok.com",
"ip_restriction": {"allow_cidrs": ["allowed"], "deny_cidrs": ["denied"]},
# "key": __file__,
"proxy_proto": "1",
# Can't be used in conjunction with "domain", but validated in other tests
# "subdomain": "pyngrok",
"terminate_at": "edge",
"traffic_policy":
{"inbound": {"name": "inbound-policy", "expressions": "inbound-policy-expression",
"actions": {"type": "inbound-policy-actions-type",
"config": "inbound-policy-actions-config"}},
"outbound": {"name": "outbound-policy", "expressions": "outbound-policy-expression",
"actions": {"type": "outbound-policy-actions-type",
"config": "outbound-policy-actions-config"}}}
}
}
}
config_path = os.path.join(self.config_dir, "config_v3_2.yml")
installer.install_default_config(config_path, config, ngrok_version="v3")
pyngrok_config = self.copy_with_updates(self.pyngrok_config_v3,
api_key="api-key",
config_path=config_path)

# WHEN
# This test ensures the validity of the config (pyngrok will fail with PyngrokNgrokError if the above is
# invalid), not that the config can start a valid tunnel (thus PyngrokNgrokHTTPError is thrown when pyngrok
# tries to open the tunnel after ngrok successfully starts, but we can ignore that for this test
with self.assertRaises(PyngrokNgrokHTTPError):
ngrok.connect(name="my-tunnel", pyngrok_config=pyngrok_config)

# def test_full_v3_tunnel_definitions(self):
# # GIVEN
# config = {
# "version": "3",
# "tunnels": {
# "my-tunnel": {
# "proto": "tcp",
# "domain": "pyngrok.com",
# "addr": "5000",
# "inspect": "false",
# "labels": ["edge=some-edge-id"],
# "basic_auth": ["auth-token"],
# "host_header": "host-header",
# "crt": "crt",
# "key": "key",
# "client_cas": "clientCas",
# "remote_addr": "remoteAddr",
# "metadata": "metadata",
# "compression": "false",
# "mutual_tls_cas": "mutualTlsCas",
# "proxy_proto": "1",
# "websocket_tcp_converter": "false",
# "terminate_at": "provider",
# "request_header": {"add": "req-addition", "remove": "req-subtraction"},
# "response_header": {"add": "res-addition", "remove": "res-subtraction"},
# "ip_restriction": {"allow_cidrs": "allowed", "deny_cidrs": "denied"},
# "verify_webhook": {"provider": "provider", "secret": "secret"},
# "user_agent_filter": {"allow": "allow-user-agent", "deny": "deny-user-agent"},
# "policy":
# {"inbound": {"name": "inbound-policy", "expressions": "inbound-policy-expression",
# "actions": {"type": "inbound-policy-actions-type",
# "config": "inbound-policy-actions-config"}}}
# }
# },
# "endpoints": {
# "name": "my-endpoint",
# "url": "https://pyngrok.com",
# "upstream": {
# "url": "8080",
# "protocol": "http1"
# },
# "metadata": "{ \"id\": \"example-app\" }",
# "description": "my endpoint",
# "traffic_policy": {
# "on_http_request": {
# "name": "foobar",
# "expressions": "'bar' in getQueryParam('foo')",
# "actions": {
# "type": "custom-response",
# "config": {
# "status_code": 404,
# "content": "not found",
# "headers": {
# "content-type": "text/plain"
# }
# }
# }
# }
# }
# }
# }
# config_path = os.path.join(self.config_dir, "config_v3_2.yml")
#
# installer.install_default_config(config_path, config, ngrok_version="v3", config_version="3")
# pyngrok_config = self.copy_with_updates(self.pyngrok_config_v3,
# api_key="api-key",
# config_path=config_path)
#
# # WHEN
# ngrok.connect(name="my-tunnel", pyngrok_config=pyngrok_config)

################################################################################
# Tests below this point don't need to start a long-lived ngrok process, they
# are asserting on pyngrok-specific code or edge cases.
Expand Down Expand Up @@ -999,42 +1205,20 @@ def test_set_auth_token_fails(self, mock_check_output):

@mock.patch('pyngrok.ngrok.api_request')
@mock.patch('pyngrok.ngrok.get_ngrok_process')
def test_full_tunnel_definitions(self, mock_get_ngrok_process, mock_api_request):
def test_config_passed_to_api_request(self, mock_get_ngrok_process, mock_api_request):
# GIVEN
config = {
"version": "2",
"tunnels": {
"my-tunnel": {
"domain": "pyngrok.com",
"addr": "5000",
"inspect": "false",
"labels": ["edge=some-edge-id"],
"basic_auth": ["auth-token"],
"host-header": "host-header",
"hostname": "hostname",
"crt": "crt",
"key": "key",
"clientCas": "clientCas",
"remoteAddr": "remoteAddr",
"metadata": "metadata",
"compression": "false",
"mutualTlsCas": "mutualTlsCas",
"proxyProto": "proxyProto",
"websocketTcpConverter": "false",
"terminateAt": "provider",
"request_header": {"add": "req-addition", "remove": "req-subtraction"},
"response_header": {"add": "res-addition", "remove": "res-subtraction"},
"ip_restriction": {"allow_cidrs": "allowed", "deny_cidrs": "denied"},
"verify_webhook": {"provider": "provider", "secret": "secret"},
"allow_user_agent": {"allow": "allow-user-agent", "deny": "deny-user-agent"},
"policy":
{"inbound": {"name": "inbound-policy", "expressions": "inbound-policy-expression",
"actions": {"type": "inbound-policy-actions-type",
"config": "inbound-policy-actions-config"}}}
"proto": "http",
"addr": "8000"
}
}
}
config_path = os.path.join(self.config_dir, "config_v3_2.yml")
installer.install_default_config(config_path, config, ngrok_version="v3")

installer.install_default_config(config_path, config, ngrok_version="v3", config_version="3")
pyngrok_config = self.copy_with_updates(self.pyngrok_config_v3,
api_key="api-key",
config_path=config_path)
Expand Down

0 comments on commit 0755638

Please sign in to comment.