From 616d3b28486d9516949003c025d8df57c358f8d1 Mon Sep 17 00:00:00 2001 From: umesh sharma Date: Wed, 22 Apr 2026 22:15:59 -0700 Subject: [PATCH 1/2] NETCONF: centralize endpoint validation + hermetic unit/regression tests --- .gitignore | 1 + sdk/python/core/tests/test_endpoint.py | 307 ++++++++++++++++++ .../tests/test_netconf_provider_config.py | 289 +++++++++++++++++ .../core/ydk/path/sessions/netconf_session.py | 4 +- sdk/python/core/ydk/providers/_endpoint.py | 245 ++++++++++++++ .../core/ydk/providers/netconf_provider.py | 4 +- 6 files changed, 846 insertions(+), 4 deletions(-) create mode 100644 sdk/python/core/tests/test_endpoint.py create mode 100644 sdk/python/core/tests/test_netconf_provider_config.py create mode 100644 sdk/python/core/ydk/providers/_endpoint.py diff --git a/.gitignore b/.gitignore index 23bc0deab..5374c8850 100644 --- a/.gitignore +++ b/.gitignore @@ -43,3 +43,4 @@ sdk/python/gnmi/dist sdk/python/gnmi/ydk_service_gnmi.egg-info venv temp +.claude/ diff --git a/sdk/python/core/tests/test_endpoint.py b/sdk/python/core/tests/test_endpoint.py new file mode 100644 index 000000000..1cca8004c --- /dev/null +++ b/sdk/python/core/tests/test_endpoint.py @@ -0,0 +1,307 @@ +# ---------------------------------------------------------------- +# Copyright 2016-2019 Cisco Systems +# +# 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. +# ------------------------------------------------------------------ + +"""test_endpoint.py + +Standalone unit tests for ``ydk.providers._endpoint``. + +These tests deliberately avoid importing ``ydk.providers`` as a +package (which would trigger import of the compiled ``ydk_`` extension +via ``netconf_provider.py``). Instead, the ``_endpoint`` module is +loaded directly from its file path so the tests run on any machine +with a Python interpreter, with no C++ build or NETCONF server. +""" + +from __future__ import absolute_import + +import os +import sys +import unittest +import importlib.util + +# Make the ydk source tree importable so `from ydk.errors import ...` +# inside _endpoint.py resolves against the source (not an installed +# package). +_HERE = os.path.dirname(os.path.abspath(__file__)) +_CORE = os.path.dirname(_HERE) # .../sdk/python/core +if _CORE not in sys.path: + sys.path.insert(0, _CORE) + +from ydk.errors import YInvalidArgumentError # noqa: E402 + +# Load _endpoint.py directly, bypassing ydk/providers/__init__.py +# (which would pull in the ydk_ compiled extension). +_ENDPOINT_PATH = os.path.join( + _CORE, "ydk", "providers", "_endpoint.py") +_spec = importlib.util.spec_from_file_location( + "ydk_providers_endpoint_under_test", _ENDPOINT_PATH) +_endpoint = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_endpoint) + +parse_endpoint = _endpoint.parse_endpoint +validate_endpoint = _endpoint.validate_endpoint + + +class TestParseEndpointValid(unittest.TestCase): + """Happy-path inputs to ``parse_endpoint``.""" + + def test_bare_hostname(self): + self.assertEqual( + parse_endpoint("router.example.com"), + (None, "router.example.com", None)) + + def test_bare_ipv4(self): + self.assertEqual( + parse_endpoint("192.0.2.1"), + (None, "192.0.2.1", None)) + + def test_missing_scheme_host_port(self): + # Explicitly: no scheme, host:port form + self.assertEqual( + parse_endpoint("host.example.com:830"), + (None, "host.example.com", 830)) + + def test_scheme_host(self): + self.assertEqual( + parse_endpoint("ssh://router1"), + ("ssh", "router1", None)) + + def test_scheme_host_port(self): + self.assertEqual( + parse_endpoint("ssh://10.0.0.1:22"), + ("ssh", "10.0.0.1", 22)) + + def test_scheme_case_insensitive(self): + self.assertEqual( + parse_endpoint("SSH://host:830"), + ("ssh", "host", 830)) + + def test_tcp_scheme(self): + self.assertEqual( + parse_endpoint("tcp://host:4334"), + ("tcp", "host", 4334)) + + def test_ipv6_bracketed_no_port(self): + self.assertEqual( + parse_endpoint("[2001:db8::1]"), + (None, "2001:db8::1", None)) + + def test_ipv6_bracketed_with_port(self): + # Explicit user-requested case: IPv6 bracket + port + self.assertEqual( + parse_endpoint("[2001:db8::1]:830"), + (None, "2001:db8::1", 830)) + + def test_ipv6_scheme_bracketed_port(self): + self.assertEqual( + parse_endpoint("ssh://[::1]:12022"), + ("ssh", "::1", 12022)) + + def test_ipv6_bare_no_brackets(self): + # Bare IPv6 with >=2 colons is treated as host-only (no port). + self.assertEqual( + parse_endpoint("2001:db8::1"), + (None, "2001:db8::1", None)) + + def test_leading_trailing_whitespace_stripped(self): + self.assertEqual( + parse_endpoint(" ssh://host:830 "), + ("ssh", "host", 830)) + + def test_fqdn_trailing_dot(self): + self.assertEqual( + parse_endpoint("router.example.com."), + (None, "router.example.com", None)) + + def test_underscore_in_hostname_label(self): + # Pragmatic allowance for lab/vendor hostnames. + self.assertEqual( + parse_endpoint("lab_router_1"), + (None, "lab_router_1", None)) + + def test_port_boundaries(self): + self.assertEqual(parse_endpoint("h:1"), (None, "h", 1)) + self.assertEqual(parse_endpoint("h:65535"), (None, "h", 65535)) + + +class TestParseEndpointInvalid(unittest.TestCase): + """Malformed inputs must raise ``YInvalidArgumentError`` + with an actionable message.""" + + def _assert_rejects(self, value, msg_substr): + with self.assertRaises(YInvalidArgumentError) as cm: + parse_endpoint(value) + self.assertIn(msg_substr, str(cm.exception)) + + def test_empty_string(self): + self._assert_rejects("", "endpoint is empty") + + def test_whitespace_only(self): + # Explicit user-requested case: whitespace host + self._assert_rejects(" ", "endpoint is empty") + + def test_non_string(self): + self._assert_rejects(None, "expected str") + self._assert_rejects(12345, "expected str") + + def test_unsupported_scheme(self): + # Explicit user-requested case + self._assert_rejects("http://host:80", "unsupported scheme") + self._assert_rejects("telnet://host", "unsupported scheme") + + def test_scheme_no_host(self): + self._assert_rejects("ssh://", "no host component") + + def test_userinfo_rejected(self): + # Explicit user-requested case: userinfo in URL + self._assert_rejects( + "ssh://admin:secret@host:830", "userinfo") + self._assert_rejects( + "user@host", "userinfo") + + def test_port_zero(self): + # Explicit user-requested case: invalid port 0 + self._assert_rejects("host:0", "out of range") + + def test_port_65536(self): + # Explicit user-requested case: invalid port 65536 + self._assert_rejects("host:65536", "out of range") + + def test_port_non_numeric(self): + # Explicit user-requested case + self._assert_rejects("host:abc", "not a positive integer") + self._assert_rejects("host:8 30", "not a positive integer") + self._assert_rejects("host:-1", "not a positive integer") + + def test_trailing_colon_no_port(self): + self._assert_rejects("host:", "trailing ':' but no port") + + def test_ipv6_unclosed_bracket(self): + self._assert_rejects("[2001:db8::1", "missing closing bracket") + self._assert_rejects( + "ssh://[2001:db8::1:830", "missing closing bracket") + + def test_ipv6_garbage_after_bracket(self): + self._assert_rejects( + "[2001:db8::1]garbage", "expected ':' after IPv6") + + def test_ipv6_empty_brackets(self): + self._assert_rejects("[]", "IPv6 literal is empty") + + def test_embedded_whitespace(self): + self._assert_rejects("ho st", "whitespace") + + def test_path_separator(self): + self._assert_rejects("host/path", "contains '/'") + self._assert_rejects("ssh://host/path", "contains '/'") + + def test_invalid_hostname_label(self): + self._assert_rejects("host..example.com", "invalid host label") + self._assert_rejects("-host", "invalid host label") + self._assert_rejects("host-", "invalid host label") + self._assert_rejects("a" * 64, "invalid host label") + + +class TestValidateEndpoint(unittest.TestCase): + """Tests for ``validate_endpoint`` (constructor-arg validation).""" + + def _assert_rejects(self, address, port, protocol, msg_substr): + with self.assertRaises(YInvalidArgumentError) as cm: + validate_endpoint(address, port, protocol) + self.assertIn(msg_substr, str(cm.exception)) + + # --- happy path / normalisation --------------------------------- + + def test_representative_scenario(self): + # Matches the default test harness: ssh://admin:admin@127.0.0.1 + # -> address="127.0.0.1", port=12022, protocol="ssh" + self.assertEqual( + validate_endpoint("127.0.0.1", 12022, "ssh"), + ("127.0.0.1", 12022)) + + def test_port_none_defaults_to_830(self): + # Must preserve existing wrapper behaviour. + self.assertEqual( + validate_endpoint("host", None, "ssh"), + ("host", 830)) + + def test_string_port_coerced(self): + self.assertEqual( + validate_endpoint("host", "22", "ssh"), + ("host", 22)) + + def test_ipv6_brackets_stripped(self): + self.assertEqual( + validate_endpoint("[::1]", 830, "ssh"), + ("::1", 830)) + + def test_whitespace_stripped(self): + self.assertEqual( + validate_endpoint(" host ", 830, "ssh"), + ("host", 830)) + + def test_protocol_case_insensitive_validation(self): + # Validated case-insensitively; value is NOT transformed + # (wrapper passes its original protocol string through). + host, port = validate_endpoint("host", 830, "SSH") + self.assertEqual((host, port), ("host", 830)) + + def test_custom_default_port(self): + self.assertEqual( + validate_endpoint("host", None, "tcp", default_port=4334), + ("host", 4334)) + + # --- rejection cases -------------------------------------------- + + def test_empty_address(self): + self._assert_rejects("", 830, "ssh", "host is empty") + + def test_none_address(self): + self._assert_rejects(None, 830, "ssh", "expected str") + + def test_uri_passed_as_address(self): + # Common user mistake: passing a full URI as `address`. + self._assert_rejects( + "ssh://host:830", 830, "ssh", "contains '/'") + + def test_port_zero_rejected(self): + self._assert_rejects("host", 0, "ssh", "out of range") + + def test_port_negative_rejected(self): + self._assert_rejects("host", -1, "ssh", "out of range") + + def test_port_too_large_rejected(self): + self._assert_rejects("host", 99999, "ssh", "out of range") + + def test_port_bool_rejected(self): + self._assert_rejects("host", True, "ssh", "got bool") + + def test_port_float_rejected(self): + self._assert_rejects("host", 830.0, "ssh", "got float") + + def test_port_non_numeric_str_rejected(self): + self._assert_rejects("host", "abc", "ssh", + "not a positive integer") + + def test_unsupported_protocol(self): + self._assert_rejects("host", 830, "https", "not supported") + + def test_non_string_protocol(self): + self._assert_rejects("host", 830, 123, "expected str") + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/sdk/python/core/tests/test_netconf_provider_config.py b/sdk/python/core/tests/test_netconf_provider_config.py new file mode 100644 index 000000000..99e54e3e3 --- /dev/null +++ b/sdk/python/core/tests/test_netconf_provider_config.py @@ -0,0 +1,289 @@ +# ---------------------------------------------------------------- +# Copyright 2016-2019 Cisco Systems +# +# 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. +# ------------------------------------------------------------------ + +"""test_netconf_provider_config.py + +Regression test: confirm that after wiring ``validate_endpoint`` into +the ``NetconfServiceProvider`` / ``NetconfSession`` wrappers, the +arguments forwarded to the underlying C++ base-class constructor are +**byte-for-byte identical** to the pre-change behaviour for valid +inputs. + +The compiled ``ydk_`` extension is replaced with stub base classes +that simply record every positional-argument tuple passed to their +``__init__``. This lets the test run on any machine with no C++ build +and no NETCONF server. +""" + +from __future__ import absolute_import + +import os +import sys +import types +import unittest +from unittest.mock import MagicMock + +# --------------------------------------------------------------------- +# 1. Make the ydk source tree importable. +# --------------------------------------------------------------------- +_HERE = os.path.dirname(os.path.abspath(__file__)) +_CORE = os.path.dirname(_HERE) # .../sdk/python/core +if _CORE not in sys.path: + sys.path.insert(0, _CORE) + +# --------------------------------------------------------------------- +# 2. Capture list for args passed to mock base __init__. +# Each entry is (class_name, args_tuple). +# --------------------------------------------------------------------- +_CAPTURED = [] + + +class _RecordingProviderBase(object): + """Stand-in for the pybind11 ``_NetconfServiceProvider``.""" + def __init__(self, *args, **kwargs): + # The real wrappers never pass kwargs to super().__init__, + # but record them anyway so a surprise kwarg fails loudly. + _CAPTURED.append(("NetconfServiceProvider", args, kwargs)) + + +class _RecordingSessionBase(object): + """Stand-in for the pybind11 ``_NetconfSession``.""" + def __init__(self, *args, **kwargs): + _CAPTURED.append(("NetconfSession", args, kwargs)) + + +# --------------------------------------------------------------------- +# 3. Inject stub modules *before* importing the wrappers. +# +# (a) `ydk_` (the compiled extension) and its submodules are +# replaced so `from ydk_.providers import NetconfServiceProvider` +# and `from ydk_.path import NetconfSession` yield our recording +# base classes. +# +# (b) `ydk.providers`, `ydk.path`, `ydk.path.sessions` are replaced +# with lightweight package stubs whose __path__ points at the +# real source directory. This BYPASSES their heavy __init__.py +# chains (which pull in codec_provider -> ydk.types -> +# ydk.ext.types -> compiled symbols) while still letting Python +# resolve submodule imports like `ydk.providers._endpoint` and +# `ydk.providers.netconf_provider` directly from disk. +# --------------------------------------------------------------------- +_ydk_underscore = MagicMock(name="ydk_") +_ydk_underscore_providers = types.ModuleType("ydk_.providers") +_ydk_underscore_providers.NetconfServiceProvider = _RecordingProviderBase +_ydk_underscore_path = types.ModuleType("ydk_.path") +_ydk_underscore_path.NetconfSession = _RecordingSessionBase + +sys.modules["ydk_"] = _ydk_underscore +sys.modules["ydk_.providers"] = _ydk_underscore_providers +sys.modules["ydk_.path"] = _ydk_underscore_path + + +def _pkg_stub(name, *path_parts): + """Insert a minimal package stub into sys.modules. + + The stub has __path__ so Python can load submodules from disk + without executing the real __init__.py. + """ + m = types.ModuleType(name) + m.__path__ = [os.path.join(_CORE, *path_parts)] + sys.modules[name] = m + return m + + +_pkg_stub("ydk.providers", "ydk", "providers") +_pkg_stub("ydk.path", "ydk", "path") +_pkg_stub("ydk.path.sessions", "ydk", "path", "sessions") + +# --------------------------------------------------------------------- +# 4. Now import the wrappers; they subclass our recording bases. +# --------------------------------------------------------------------- +from ydk.providers.netconf_provider import NetconfServiceProvider # noqa: E402 +from ydk.path.sessions.netconf_session import NetconfSession # noqa: E402 +from ydk.errors import YInvalidArgumentError # noqa: E402 + + +# --------------------------------------------------------------------- +# 5. Representative scenario (mirrors the test harness default: +# ssh://admin:admin@127.0.0.1:12022 parsed by urllib). +# --------------------------------------------------------------------- +ADDRESS = "127.0.0.1" +USERNAME = "admin" +PASSWORD = "admin" +PORT = 12022 +PROTOCOL = "ssh" +# Defaults, made explicit so the expected tuples below are obvious: +ON_DEMAND = True +COMMON_CACHE = False +TIMEOUT = -1 # wrapper normalises None -> -1 +PRIV_KEY = "/tmp/id_rsa" +PUB_KEY = "/tmp/id_rsa.pub" +_REPO_SENTINEL = object() # any non-None object triggers repo branch + + +class NetconfProviderConfigRegressionTest(unittest.TestCase): + """Assert the args forwarded to ``super().__init__`` are unchanged. + + "Unchanged" means: for valid input, the positional tuple the + wrapper passes to its C++ base class's ``__init__`` is the same + tuple the pre-patch code would have passed. The pre-patch code + did ``if port is None: port = 830`` and otherwise forwarded raw + values; ``validate_endpoint`` does the same for valid inputs and + only adds rejection of *invalid* inputs. + """ + + def setUp(self): + _CAPTURED[:] = [] + + def _last(self, expect_class): + self.assertEqual(len(_CAPTURED), 1, + "expected exactly one base.__init__ call") + cls, args, kwargs = _CAPTURED[0] + self.assertEqual(cls, expect_class) + self.assertEqual(kwargs, {}) + return args + + # --- NetconfServiceProvider ------------------------------------ + + def test_provider_branch1_no_repo_password(self): + """repo=None, password auth -> the representative scenario.""" + NetconfServiceProvider( + ADDRESS, USERNAME, PASSWORD, PORT, PROTOCOL, + ON_DEMAND, COMMON_CACHE, timeout=None) + # Pre-patch code path (netconf_provider.py original lines 53-54): + # self.nsp.__init__(address, username, password, port, + # protocol, on_demand, common_cache, timeout) + self.assertEqual( + self._last("NetconfServiceProvider"), + (ADDRESS, USERNAME, PASSWORD, PORT, + PROTOCOL, ON_DEMAND, COMMON_CACHE, TIMEOUT)) + + def test_provider_branch2_no_repo_key_auth(self): + NetconfServiceProvider( + ADDRESS, USERNAME, + private_key_path=PRIV_KEY, public_key_path=PUB_KEY, + port=PORT) + # Pre-patch (original lines 56-58): + # self.nsp.__init__(address, username, + # private_key_path, public_key_path, + # port, on_demand, common_cache, timeout) + self.assertEqual( + self._last("NetconfServiceProvider"), + (ADDRESS, USERNAME, PRIV_KEY, PUB_KEY, + PORT, ON_DEMAND, COMMON_CACHE, TIMEOUT)) + + def test_provider_branch3_repo_password(self): + NetconfServiceProvider( + ADDRESS, USERNAME, PASSWORD, PORT, PROTOCOL, + repo=_REPO_SENTINEL) + # Pre-patch (original lines 61-62): + # self.nsp.__init__(repo, address, username, password, + # port, protocol, on_demand, timeout) + self.assertEqual( + self._last("NetconfServiceProvider"), + (_REPO_SENTINEL, ADDRESS, USERNAME, PASSWORD, + PORT, PROTOCOL, ON_DEMAND, TIMEOUT)) + + def test_provider_branch4_repo_key_auth(self): + NetconfServiceProvider( + ADDRESS, USERNAME, + private_key_path=PRIV_KEY, public_key_path=PUB_KEY, + port=PORT, repo=_REPO_SENTINEL) + # Pre-patch (original lines 64-66): + # self.nsp.__init__(repo, address, username, + # private_key_path, public_key_path, + # port, on_demand, timeout) + self.assertEqual( + self._last("NetconfServiceProvider"), + (_REPO_SENTINEL, ADDRESS, USERNAME, + PRIV_KEY, PUB_KEY, PORT, ON_DEMAND, TIMEOUT)) + + def test_provider_port_none_defaults_to_830(self): + """Behavioural parity: port=None -> 830 (was a plain `if`).""" + NetconfServiceProvider(ADDRESS, USERNAME, PASSWORD, port=None) + args = self._last("NetconfServiceProvider") + self.assertEqual(args[3], 830) + + def test_provider_invalid_address_raises_before_base_init(self): + """New behaviour: validation error fires before C++ layer.""" + with self.assertRaises(YInvalidArgumentError) as cm: + NetconfServiceProvider("", USERNAME, PASSWORD, PORT) + self.assertIn("host is empty", str(cm.exception)) + self.assertEqual(_CAPTURED, [], + "base __init__ must not be reached") + + # --- NetconfSession -------------------------------------------- + + def test_session_branch1_no_repo_password(self): + NetconfSession( + ADDRESS, USERNAME, PASSWORD, PORT, PROTOCOL, + ON_DEMAND, COMMON_CACHE, timeout=None) + # Pre-patch (netconf_session.py original lines 52-54): + # self.ns.__init__(address, username, password, + # port, protocol, on_demand, + # common_cache, timeout) + self.assertEqual( + self._last("NetconfSession"), + (ADDRESS, USERNAME, PASSWORD, PORT, + PROTOCOL, ON_DEMAND, COMMON_CACHE, TIMEOUT)) + + def test_session_branch2_no_repo_key_auth(self): + NetconfSession( + ADDRESS, USERNAME, + private_key_path=PRIV_KEY, public_key_path=PUB_KEY, + port=PORT) + # Pre-patch (original lines 56-59): + # self.ns.__init__(address, username, + # private_key_path, public_key_path, + # port, protocol, on_demand, + # common_cache, timeout) + self.assertEqual( + self._last("NetconfSession"), + (ADDRESS, USERNAME, PRIV_KEY, PUB_KEY, + PORT, PROTOCOL, ON_DEMAND, COMMON_CACHE, TIMEOUT)) + + def test_session_branch3_repo_password(self): + NetconfSession( + ADDRESS, USERNAME, PASSWORD, PORT, PROTOCOL, + repo=_REPO_SENTINEL) + # Pre-patch (original lines 62-64): + # self.ns.__init__(repo, address, username, password, + # port, protocol, + # on_demand, timeout) + self.assertEqual( + self._last("NetconfSession"), + (_REPO_SENTINEL, ADDRESS, USERNAME, PASSWORD, + PORT, PROTOCOL, ON_DEMAND, TIMEOUT)) + + def test_session_branch4_repo_key_auth(self): + NetconfSession( + ADDRESS, USERNAME, + private_key_path=PRIV_KEY, public_key_path=PUB_KEY, + port=PORT, repo=_REPO_SENTINEL) + # Pre-patch (original lines 66-69): + # self.ns.__init__(repo, address, username, + # private_key_path, public_key_path, + # port, protocol, on_demand, + # common_cache, timeout) + self.assertEqual( + self._last("NetconfSession"), + (_REPO_SENTINEL, ADDRESS, USERNAME, + PRIV_KEY, PUB_KEY, PORT, PROTOCOL, + ON_DEMAND, COMMON_CACHE, TIMEOUT)) + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/sdk/python/core/ydk/path/sessions/netconf_session.py b/sdk/python/core/ydk/path/sessions/netconf_session.py index 0e9626bea..d1718d9c2 100644 --- a/sdk/python/core/ydk/path/sessions/netconf_session.py +++ b/sdk/python/core/ydk/path/sessions/netconf_session.py @@ -25,6 +25,7 @@ """ from ydk_.path import NetconfSession as _NetconfSession +from ydk.providers._endpoint import validate_endpoint class NetconfSession(_NetconfSession): @@ -39,8 +40,7 @@ def __init__(self, if timeout is None: timeout = -1 - if port is None: - port = 830 + address, port = validate_endpoint(address, port, protocol) if private_key_path is None: private_key_path = "" if public_key_path is None: diff --git a/sdk/python/core/ydk/providers/_endpoint.py b/sdk/python/core/ydk/providers/_endpoint.py new file mode 100644 index 000000000..8ba19fbee --- /dev/null +++ b/sdk/python/core/ydk/providers/_endpoint.py @@ -0,0 +1,245 @@ +# ---------------------------------------------------------------- +# Copyright 2016-2019 Cisco Systems +# +# 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. +# ------------------------------------------------------------------ + +""" +_endpoint.py + +Internal helpers for parsing and validating connection endpoints +(scheme / host / port) used by service providers and sessions. + +This module deliberately has **no** dependency on the compiled +``ydk_`` extension so it can be unit-tested standalone. +""" + +from ydk.errors import YInvalidArgumentError + + +_NETCONF_DEFAULT_PORT = 830 +_SUPPORTED_SCHEMES = frozenset(("ssh", "tcp")) + +# Hostname label characters. RFC 1123 permits letters, digits and +# hyphens; underscore is added here pragmatically because many lab +# and vendor devices use it in hostnames. +_LABEL_CHARS = frozenset( + "abcdefghijklmnopqrstuvwxyz" + "ABCDEFGHIJKLMNOPQRSTUVWXYZ" + "0123456789-_" +) + + +def _is_valid_label(label): + if not 1 <= len(label) <= 63: + return False + if label[0] == "-" or label[-1] == "-": + return False + return all(ch in _LABEL_CHARS for ch in label) + + +def _looks_like_ipv6(host): + # IPv6 literals (after bracket stripping) contain at least one ':'. + return ":" in host + + +def _coerce_port(raw, context): + """Convert *raw* to a valid int port or raise. + + Accepts ``int`` (excluding ``bool``) and all-digit ``str``. + Rejects out-of-range (<1 or >65535), floats and other types. + *context* is prepended to error messages so callers can tell + which input was at fault (e.g. ``"port"`` vs ``"parse_endpoint"``). + """ + if isinstance(raw, bool): + raise YInvalidArgumentError( + "%s: expected int port, got bool" % context) + if isinstance(raw, int): + port = raw + elif isinstance(raw, str): + stripped = raw.strip() + if not stripped or not stripped.isdigit(): + raise YInvalidArgumentError( + "%s: port %r is not a positive integer" % (context, raw)) + port = int(stripped) + else: + raise YInvalidArgumentError( + "%s: expected int or numeric str port, got %s" + % (context, type(raw).__name__)) + if not 1 <= port <= 65535: + raise YInvalidArgumentError( + "%s: port %d out of range 1..65535" % (context, port)) + return port + + +def _coerce_host(raw, context): + """Normalise a host string (hostname / IPv4 / IPv6 literal). + + Strips surrounding whitespace, IPv6 brackets, and a trailing FQDN + dot. Rejects empty strings, embedded whitespace, path separators + and malformed hostname labels. Returns the normalised host. + """ + if not isinstance(raw, str): + raise YInvalidArgumentError( + "%s: expected str host, got %s" + % (context, type(raw).__name__)) + host = raw.strip() + if not host: + raise YInvalidArgumentError("%s: host is empty" % context) + if any(ch.isspace() for ch in host): + raise YInvalidArgumentError( + "%s: host %r contains whitespace" % (context, raw)) + if host.startswith("["): + if not host.endswith("]"): + raise YInvalidArgumentError( + "%s: IPv6 literal %r missing closing bracket" + % (context, raw)) + host = host[1:-1] + if not host: + raise YInvalidArgumentError( + "%s: IPv6 literal is empty" % context) + if "/" in host: + raise YInvalidArgumentError( + "%s: host %r contains '/'; pass a bare host, not a URI" + % (context, raw)) + if _looks_like_ipv6(host): + if ":::" in host or not host.replace(":", ""): + raise YInvalidArgumentError( + "%s: malformed IPv6 literal %r" % (context, raw)) + return host + if host.endswith("."): + host = host[:-1] + if not host: + raise YInvalidArgumentError("%s: host is empty" % context) + for label in host.split("."): + if not _is_valid_label(label): + raise YInvalidArgumentError( + "%s: invalid host label %r in %r" + % (context, label, raw)) + return host + + +def parse_endpoint(endpoint): + """Parse a scheme://host[:port] URI or a bare host[:port] string. + + Accepted forms (scheme is case-insensitive, optional):: + + ssh://host + ssh://host:830 + ssh://[2001:db8::1]:830 + tcp://router.example.com + host.example.com + host.example.com:830 + [2001:db8::1]:830 + 2001:db8::1 (bare IPv6, no port) + + Returns ``(scheme, host, port)`` where *scheme* and *port* are + ``None`` if absent; IPv6 brackets are stripped from *host*. + + Credentials (``user:pass@``) are **not** supported and are + rejected explicitly -- callers pass them as separate parameters. + + Raises ``YInvalidArgumentError`` on any malformed input. + """ + if not isinstance(endpoint, str): + raise YInvalidArgumentError( + "parse_endpoint: expected str, got %s" + % type(endpoint).__name__) + ep = endpoint.strip() + if not ep: + raise YInvalidArgumentError("parse_endpoint: endpoint is empty") + + scheme = None + rest = ep + if "://" in ep: + scheme_raw, _, rest = ep.partition("://") + scheme = scheme_raw.lower() + if scheme not in _SUPPORTED_SCHEMES: + raise YInvalidArgumentError( + "parse_endpoint: unsupported scheme %r " + "(expected one of %s)" + % (scheme_raw, sorted(_SUPPORTED_SCHEMES))) + if not rest: + raise YInvalidArgumentError( + "parse_endpoint: %r has no host component" % endpoint) + + if "@" in rest: + raise YInvalidArgumentError( + "parse_endpoint: userinfo (user:pass@) is not supported " + "in the address; pass credentials as separate parameters") + + host_part = rest + port_part = None + + if host_part.startswith("["): + close = host_part.find("]") + if close < 0: + raise YInvalidArgumentError( + "parse_endpoint: IPv6 literal in %r " + "missing closing bracket" % endpoint) + bracketed = host_part[:close + 1] + remainder = host_part[close + 1:] + if remainder: + if remainder[0] != ":": + raise YInvalidArgumentError( + "parse_endpoint: expected ':' after IPv6 " + "bracket in %r" % endpoint) + port_part = remainder[1:] + host_part = bracketed + elif host_part.count(":") == 1: + host_part, _, port_part = host_part.partition(":") + # else: zero colons -> bare host, no port + # >=2 colons -> bare IPv6 literal without brackets, no port + + host = _coerce_host(host_part, "parse_endpoint") + port = None + if port_part is not None: + if port_part == "": + raise YInvalidArgumentError( + "parse_endpoint: %r has trailing ':' but no port" + % endpoint) + port = _coerce_port(port_part, "parse_endpoint") + + return scheme, host, port + + +def validate_endpoint(address, port, protocol, + default_port=_NETCONF_DEFAULT_PORT): + """Validate and normalise NETCONF constructor endpoint arguments. + + :param address: host string (hostname, IPv4, or IPv6 literal). + Brackets/whitespace/trailing-dot are stripped. + :param port: ``int`` or numeric ``str``; ``None`` falls back + to *default_port* (matching existing behaviour). + :param protocol: ``"ssh"`` or ``"tcp"`` (case-insensitive). The + value is validated but not transformed -- the + caller passes its original ``protocol`` string + to the underlying C++ layer unchanged. + :returns: ``(host, port)`` normalised values. + :raises YInvalidArgumentError: on any invalid input. + """ + host = _coerce_host(address, "address") + + if protocol is not None: + if not isinstance(protocol, str): + raise YInvalidArgumentError( + "protocol: expected str, got %s" + % type(protocol).__name__) + if protocol.lower() not in _SUPPORTED_SCHEMES: + raise YInvalidArgumentError( + "protocol: %r is not supported (expected one of %s)" + % (protocol, sorted(_SUPPORTED_SCHEMES))) + + if port is None: + return host, int(default_port) + return host, _coerce_port(port, "port") diff --git a/sdk/python/core/ydk/providers/netconf_provider.py b/sdk/python/core/ydk/providers/netconf_provider.py index bda7c1a29..2cc699fe4 100644 --- a/sdk/python/core/ydk/providers/netconf_provider.py +++ b/sdk/python/core/ydk/providers/netconf_provider.py @@ -26,6 +26,7 @@ """ from ydk_.providers import NetconfServiceProvider as _NetconfServiceProvider +from ._endpoint import validate_endpoint class NetconfServiceProvider(_NetconfServiceProvider): @@ -40,8 +41,7 @@ def __init__(self, if timeout is None: timeout = -1 - if port is None: - port = 830 + address, port = validate_endpoint(address, port, protocol) if private_key_path is None: private_key_path = "" if public_key_path is None: From 0805ab34507b58e841942e3598d6ff7afb01c7ca Mon Sep 17 00:00:00 2001 From: umesh sharma Date: Wed, 22 Apr 2026 22:23:38 -0700 Subject: [PATCH 2/2] NETCONF: centralize endpoint validation + hermetic unit/regression tests --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index 5374c8850..277f321a9 100644 --- a/.gitignore +++ b/.gitignore @@ -44,3 +44,4 @@ sdk/python/gnmi/ydk_service_gnmi.egg-info venv temp .claude/ +.claude/