From c56f6f6ad819a4fced201eab5cdaca67e8d23d85 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 29 Jan 2026 02:23:11 +0000 Subject: [PATCH 1/5] Modernize Python codebase to 3.10+ with comprehensive tooling This major update modernizes the netlib codebase to align with current Python best practices: ## Core Changes - Upgrade minimum Python version to 3.10+ - Remove all Python 2.x support and legacy code - Add full type hints throughout the codebase (PEP 484) - Implement Pydantic v2 for robust input/output validation ## Package Management - Migrate from setuptools to Poetry - Remove setup.py and requirements.txt - Add pyproject.toml with complete project configuration - Include py.typed marker for PEP 561 compliance ## Testing & Quality Assurance - Replace unittest with pytest framework - Add comprehensive test suite with 7 test modules - Implement pytest-cov for coverage reporting - Add pytest-mock for better mocking support ## Linting & Type Checking - Add ruff for fast, modern linting and formatting - Configure mypy for strict type checking - Add pylint for additional code quality checks - Create .ruff.toml configuration file ## CI/CD - Remove Travis CI (.travis.yml) - Implement GitHub Actions workflow - Add multi-version testing (Python 3.10, 3.11, 3.12) - Include automated linting, type checking, and testing - Add codecov integration for coverage reporting ## Code Improvements - Add Pydantic models for connection configurations - Modernize class syntax (remove explicit object inheritance) - Use modern string formatting (f-strings) - Add comprehensive docstrings - Improve error handling and type safety ## Documentation - Complete README.md rewrite with modern badges - Add detailed API reference - Include migration guide from v0.1.x - Add development setup instructions - Document all new features and tools Version bumped to 0.2.0 to reflect breaking changes. https://claude.ai/code/session_019D1TUrg1xPGse1HsS8pgqQ --- .github/workflows/ci.yml | 104 +++++++++++ .gitignore | 146 +++++++++++++++- .ruff.toml | 46 +++++ .travis.yml | 19 -- README.md | 333 ++++++++++++++++++++++++----------- netlib/__init__.py | 14 +- netlib/conn_type.py | 369 ++++++++++++++++++++++++++++----------- netlib/models.py | 73 ++++++++ netlib/user_keyring.py | 125 ++++++++----- py.typed | 2 + pyproject.toml | 142 +++++++++++++++ requirements.txt | 4 - scripts/tests.py | 36 ---- setup.py | 23 --- tests/__init__.py | 1 + tests/conftest.py | 20 +++ tests/test_imports.py | 49 ++++++ tests/test_keyring.py | 178 +++++++++++++++++++ tests/test_models.py | 157 +++++++++++++++++ tests/test_ssh.py | 200 +++++++++++++++++++++ tests/test_telnet.py | 172 ++++++++++++++++++ 21 files changed, 1881 insertions(+), 332 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 .ruff.toml delete mode 100644 .travis.yml create mode 100644 netlib/models.py create mode 100644 py.typed create mode 100644 pyproject.toml delete mode 100644 requirements.txt delete mode 100755 scripts/tests.py delete mode 100644 setup.py create mode 100644 tests/__init__.py create mode 100644 tests/conftest.py create mode 100644 tests/test_imports.py create mode 100644 tests/test_keyring.py create mode 100644 tests/test_models.py create mode 100644 tests/test_ssh.py create mode 100644 tests/test_telnet.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..1b63538 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,104 @@ +name: CI + +on: + push: + branches: [main, master, develop] + pull_request: + branches: [main, master, develop] + workflow_dispatch: + +jobs: + lint: + name: Lint and Type Check + runs-on: ubuntu-latest + steps: + - name: Check out code + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.12' + + - name: Install Poetry + uses: snok/install-poetry@v1 + with: + version: latest + virtualenvs-create: true + virtualenvs-in-project: true + + - name: Load cached venv + id: cached-poetry-dependencies + uses: actions/cache@v4 + with: + path: .venv + key: venv-${{ runner.os }}-${{ hashFiles('**/poetry.lock') }} + + - name: Install dependencies + if: steps.cached-poetry-dependencies.outputs.cache-hit != 'true' + run: poetry install --no-interaction --no-root + + - name: Install project + run: poetry install --no-interaction + + - name: Run ruff format check + run: poetry run ruff format --check . + + - name: Run ruff lint + run: poetry run ruff check . + + - name: Run mypy + run: poetry run mypy netlib + + - name: Run pylint + run: poetry run pylint netlib + + test: + name: Test on Python ${{ matrix.python-version }} + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ['3.10', '3.11', '3.12'] + + steps: + - name: Check out code + uses: actions/checkout@v4 + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Install Poetry + uses: snok/install-poetry@v1 + with: + version: latest + virtualenvs-create: true + virtualenvs-in-project: true + + - name: Load cached venv + id: cached-poetry-dependencies + uses: actions/cache@v4 + with: + path: .venv + key: venv-${{ runner.os }}-${{ matrix.python-version }}-${{ hashFiles('**/poetry.lock') }} + + - name: Install dependencies + if: steps.cached-poetry-dependencies.outputs.cache-hit != 'true' + run: poetry install --no-interaction --no-root + + - name: Install project + run: poetry install --no-interaction + + - name: Run tests + run: poetry run pytest + + - name: Upload coverage reports to Codecov + if: matrix.python-version == '3.12' + uses: codecov/codecov-action@v4 + with: + file: ./coverage.xml + fail_ci_if_error: false + env: + CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }} diff --git a/.gitignore b/.gitignore index 7c90419..8dec0fc 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,144 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class *.pyc -build/* -dist/* -*.egg-info -env/* + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +pip-wheel-metadata/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +.python-version + +# pipenv +Pipfile.lock + +# PEP 582 +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# Ruff +.ruff_cache/ + +# Poetry +poetry.lock + +# IDEs +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# OS +.DS_Store +Thumbs.db + +# Test files test.py diff --git a/.ruff.toml b/.ruff.toml new file mode 100644 index 0000000..c8f43e4 --- /dev/null +++ b/.ruff.toml @@ -0,0 +1,46 @@ +# Ruff configuration +# See https://docs.astral.sh/ruff/ + +line-length = 100 +target-version = "py310" + +[lint] +select = [ + "E", # pycodestyle errors + "W", # pycodestyle warnings + "F", # pyflakes + "I", # isort + "N", # pep8-naming + "UP", # pyupgrade + "B", # flake8-bugbear + "C4", # flake8-comprehensions + "DTZ", # flake8-datetimez + "T10", # flake8-debugger + "EM", # flake8-errmsg + "ISC", # flake8-implicit-str-concat + "ICN", # flake8-import-conventions + "PIE", # flake8-pie + "PT", # flake8-pytest-style + "Q", # flake8-quotes + "RSE", # flake8-raise + "RET", # flake8-return + "SIM", # flake8-simplify + "TID", # flake8-tidy-imports + "ARG", # flake8-unused-arguments + "PTH", # flake8-use-pathlib + "PL", # pylint + "RUF", # ruff-specific rules +] + +ignore = [ + "PLR0913", # Too many arguments + "PLR2004", # Magic value used in comparison +] + +[lint.per-file-ignores] +"tests/*" = ["ARG", "PLR2004"] + +[format] +quote-style = "double" +indent-style = "space" +line-ending = "auto" diff --git a/.travis.yml b/.travis.yml deleted file mode 100644 index 42837aa..0000000 --- a/.travis.yml +++ /dev/null @@ -1,19 +0,0 @@ -language: python - -matrix: - include: - - os: linux - sudo: required - python: 2.7 - - os: linux - sudo: required - python: 3.5 - - os: linux - sudo: required - python: 3.6 - -install: - - "python setup.py install" - -script: - - "python scripts/tests.py" diff --git a/README.md b/README.md index 3580ccb..c32288b 100644 --- a/README.md +++ b/README.md @@ -1,149 +1,286 @@ # NetLib -[![Build Status](https://travis-ci.org/netopsio/netlib.svg)](https://travis-ci.org/netopsio/netlib) +[![CI](https://github.com/netopsio/netlib/actions/workflows/ci.yml/badge.svg)](https://github.com/netopsio/netlib/actions/workflows/ci.yml) +[![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/downloads/) +[![Code style: ruff](https://img.shields.io/badge/code%20style-ruff-000000.svg)](https://github.com/astral-sh/ruff) +[![Type checked: mypy](https://img.shields.io/badge/type%20checked-mypy-blue.svg)](http://mypy-lang.org/) -Netlib is an attempt at re-writing -['pyRouterLib'](https://github.com/jtdub/pyRouterLib). The goal is to create a -library that is much more efficient and easier to use to establish SSH and Telnet -connections to network devices, such as routers and switches. +NetLib is a modern Python library for establishing SSH and Telnet connections to network devices such as routers and switches. It provides a clean, type-safe API with comprehensive input validation using Pydantic. -## Install +## Features -Install the Python library. +- โœจ **Modern Python**: Built for Python 3.10+ with full type hints +- ๐Ÿ”’ **Type Safety**: Comprehensive type checking with mypy +- โœ… **Input Validation**: Pydantic models for robust input/output validation +- ๐Ÿงช **Well Tested**: Extensive test coverage with pytest +- ๐Ÿ“ฆ **Poetry**: Modern dependency management +- ๐Ÿ” **Linting**: Code quality ensured with ruff, mypy, and pylint +- ๐Ÿš€ **CI/CD**: Automated testing with GitHub Actions -``` -git clone https://github.com/jtdub/netlib.git -cd netlib +## Requirements -# For Python2 Support: -sudo python setup.py install +- Python 3.10 or higher +- Poetry (for development) -# For Python3 Support: -sudo python3 setup.py install -``` +## Installation -## Access via Telnet and SSH +### Using pip -Currently, the SSH and Telnet modules have been created. Both modules have a -very similar API structure, with the biggest difference between the two are how -connections are established to network devices. +```bash +pip install netlib +``` -To use either the SSH or Telnet module, you need to import the library into your script: +### Using Poetry -``` -from netlib.conn_type import SSH -from netlib.conn_type import Telnet +```bash +poetry add netlib ``` -From there, you define your connection parameters: +### From Source -``` -telnet = Telnet('somerouter', 'username', 'password') -ssh = SSH('somerouter', 'username', 'password') +```bash +git clone https://github.com/netopsio/netlib.git +cd netlib +poetry install ``` -Once the basic parameters have been set, you establish a connection to the -device. +## Quick Start -``` -telnet.connect() -ssh.connect() -``` +### SSH Connection -Once you are connected, you are free to send commands to you network device. If -you intend to iterate through output that is long, then you can disable paging -on the output. +```python +from netlib import SSH -``` -telnet.disable_paging() +# Create SSH connection +ssh = SSH('router.example.com', 'admin', 'password') + +# Connect to device +ssh.connect() + +# Disable paging for long output ssh.disable_paging() -telnet.command('show version') -ssh.command('show version') -``` +# Execute commands +output = ssh.command('show version') +print(output) -If you need to enter a privileged mode, you can use the set_enable api. +# Execute multiple commands +outputs = ssh.commands(['show version', 'show interfaces']) -``` -telnet.set_enable('supersecretpassword') -ssh.set_enable('supersecretpassword') +# Enter privileged mode +ssh.set_enable('enable_password') + +# Close connection +ssh.close() ``` -When you've completed your task on the device, you can close your connections. +### Telnet Connection -``` +```python +from netlib import Telnet + +# Create Telnet connection +telnet = Telnet('switch.example.com', 'admin', 'password') + +# Connect to device +telnet.connect() + +# Disable paging +telnet.disable_paging() + +# Execute commands +output = telnet.command('show version') +print(output.decode('utf-8')) + +# Close connection telnet.close() -ssh.close() ``` -At this point, those are the features that both libraries share. As the telnet -library and ssh library vary on how they parse data, there is a need for extra -functionality on the ssh library. Here is the API functionality that is -specific to the ssh library: +### Credential Management with KeyRing -``` -ssh.clear_buffer() -``` +NetLib provides secure credential storage using your operating system's native keyring (Keychain on macOS, Credential Manager on Windows, etc.): -The SSH library stores output into a buffer. Sometimes this buffer can present -results that aren't expected. Clearing the buffer should mitigate the -unexpected results. +```python +from netlib import KeyRing -## User Credentials +# Initialize KeyRing with username +keyring = KeyRing(username='admin') -When working with a large number of devices, it's inconvenient to have to type -your credentials in a large number of times and storing your credentials -directly into a script can be insecure. Therefore, I created a library that -allows you to store them, securely utilizing the 'keyring' python library. -'keyring' utilizes your operating systems native method for storing passwords. -For example, in MacOS X, the Keychain utility is utilized. +# Get credentials (will prompt to create if they don't exist) +creds = keyring.get_creds() +# Returns: {'username': 'admin', 'password': 'user_password', 'enable': 'enable_password'} -The past methods of password storage, simple_creds and simple_yaml have been depricated -and removed from netlib. +# Use with SSH/Telnet +ssh = SSH('router.example.com', creds['username'], creds['password']) +ssh.connect() +ssh.set_enable(creds['enable']) -Utilizing the keyring is simple. Import the module: +# Update credentials +keyring.set_creds() -``` ->>> from netlib.user_keyring import KeyRing +# Delete credentials +keyring.del_creds() ``` -Asign it to a variable, calling your username: -``` ->>> user = KeyRing(username='jtdub') -``` +## API Reference -If there are no credentials for the keyring, the get_creds() method will call the set_creds() -method: +### SSH Class + +**Constructor**: +```python +SSH(device_name: str, username: str, password: str, + buffer: int = 65535, delay: float = 1.0, port: int = 22) ``` ->>> user.get_creds() -No credentials keyring exist. Creating new credentials. -Enter your user password: -Confirm your user password: -Enter your enable password: -Confirm your enable password: + +**Methods**: +- `connect()` - Establish SSH connection +- `close()` - Close the connection +- `command(command: str) -> str` - Execute a single command +- `commands(commands_list: list[str] | str) -> str` - Execute multiple commands +- `disable_paging(command: str = 'term len 0')` - Disable output paging +- `set_enable(enable_password: str) -> str` - Enter privileged mode +- `clear_buffer() -> str | None` - Clear the receive buffer + +### Telnet Class + +**Constructor**: +```python +Telnet(device_name: str, username: str, password: str, + delay: float = 2.0, port: int = 23) ``` -Otherwise, the creds will be pulled from the keyring: +**Methods**: +- `connect()` - Establish Telnet connection +- `close()` - Close the connection +- `command(command: str) -> bytes` - Execute a single command +- `commands(commands_list: list[str] | str) -> str` - Execute multiple commands +- `disable_paging(command: str = 'term len 0') -> bytes` - Disable output paging +- `set_enable(enable_password: str) -> str` - Enter privileged mode + +### KeyRing Class + +**Constructor**: +```python +KeyRing(username: str) ``` ->>> user.get_creds() -{'username': 'jtdub', 'enable': u'enablepass', 'password': u'testpass'} + +**Methods**: +- `get_creds() -> dict[str, str]` - Retrieve credentials from keyring +- `set_creds()` - Set or update credentials +- `del_creds()` - Delete credentials from keyring + +## Development + +### Setup Development Environment + +```bash +# Clone the repository +git clone https://github.com/netopsio/netlib.git +cd netlib + +# Install dependencies +poetry install + +# Activate virtual environment +poetry shell ``` -The set_creds() method can be called directly. It will over-write existing creds if they exist: +### Running Tests + +```bash +# Run all tests +poetry run pytest + +# Run with coverage +poetry run pytest --cov=netlib + +# Run specific test file +poetry run pytest tests/test_ssh.py ``` ->>> user.set_creds() -Enter your user password: -Confirm your user password: -Enter your enable password: -Confirm your enable password: ->>> user.get_creds() -{'username': 'jtdub', 'enable': u'newenable', 'password': u'newpass'} + +### Code Quality + +```bash +# Format code +poetry run ruff format . + +# Lint code +poetry run ruff check . + +# Type check +poetry run mypy netlib + +# Run pylint +poetry run pylint netlib + +# Run all checks +poetry run ruff format --check . && \ +poetry run ruff check . && \ +poetry run mypy netlib && \ +poetry run pylint netlib ``` -Of course, the keyring can be deleted all together utilizing the del_creds() method: +## What's New in v0.2.0 + +- ๐ŸŽฏ **Python 3.10+ Only**: Removed Python 2.x support +- ๐Ÿ”’ **Type Hints**: Full type annotations throughout the codebase +- โœ… **Pydantic Validation**: Input/output validation using Pydantic models +- ๐Ÿ“ฆ **Poetry**: Migrated from setuptools to Poetry for package management +- ๐Ÿงช **Pytest**: Comprehensive test suite with high coverage +- ๐Ÿ” **Modern Linting**: ruff, mypy, and pylint integration +- ๐Ÿš€ **GitHub Actions**: Automated CI/CD pipeline (replacing Travis CI) +- ๐Ÿ“š **Better Documentation**: Improved API documentation and examples + +## Migration from v0.1.x + +The API remains largely compatible with v0.1.x, but there are some important changes: + +1. **Python Version**: Python 3.10+ is now required +2. **Type Safety**: All methods now have type hints +3. **Validation**: Invalid inputs will now raise Pydantic `ValidationError` +4. **Installation**: Use Poetry or pip (no more `setup.py install`) + +### Example Migration + +**Before (v0.1.x)**: +```python +ssh = SSH('router', 'admin', 'pass', buffer="8192", delay="2") ``` ->>> user.del_creds() -Enter your user password: -Deleting keyring credentials for jtdub ->>> + +**After (v0.2.0)**: +```python +# Still works! String values are automatically converted +ssh = SSH('router', 'admin', 'pass', buffer="8192", delay="2") + +# But you can now use proper types +ssh = SSH('router', 'admin', 'pass', buffer=8192, delay=2.0) ``` + +## Contributing + +Contributions are welcome! Please feel free to submit a Pull Request. + +1. Fork the repository +2. Create your feature branch (`git checkout -b feature/amazing-feature`) +3. Make your changes +4. Run tests and linting (`poetry run pytest && poetry run ruff check .`) +5. Commit your changes (`git commit -m 'Add amazing feature'`) +6. Push to the branch (`git push origin feature/amazing-feature`) +7. Open a Pull Request + +## License + +This project is licensed under the MIT License - see the LICENSE file for details. + +## Credits + +NetLib is a rewrite of [pyRouterLib](https://github.com/jtdub/pyRouterLib) with a focus on modern Python practices, type safety, and maintainability. + +## Author + +James Williams + +## Links + +- **GitHub**: https://github.com/netopsio/netlib +- **Issues**: https://github.com/netopsio/netlib/issues diff --git a/netlib/__init__.py b/netlib/__init__.py index 5ce2f56..732a353 100644 --- a/netlib/__init__.py +++ b/netlib/__init__.py @@ -1,7 +1,9 @@ -from __future__ import division -from __future__ import absolute_import -from __future__ import print_function -from __future__ import unicode_literals +"""NetLib - Network device connection library.""" -__name__ = 'netlib' -__version__ = '0.1.0' +__name__ = "netlib" +__version__ = "0.2.0" + +from netlib.conn_type import SSH, Telnet +from netlib.user_keyring import KeyRing + +__all__ = ["SSH", "Telnet", "KeyRing"] diff --git a/netlib/conn_type.py b/netlib/conn_type.py index 605b2d9..2509600 100644 --- a/netlib/conn_type.py +++ b/netlib/conn_type.py @@ -1,136 +1,307 @@ -class SSH(object): - - def __init__(self, device_name, username, password, buffer="65535", - delay="1", port="22"): - import paramiko - import time - import re - self.paramiko = paramiko - self.time = time - self.re = re - self.device_name = device_name - self.username = username - self.password = password - self.buffer = int(buffer) - self.delay = int(delay) - self.port = int(port) - - def connect(self): - self.pre_conn = self.paramiko.SSHClient() - self.pre_conn.set_missing_host_key_policy( - self.paramiko.AutoAddPolicy()) - self.pre_conn.connect(self.device_name, username=self.username, - password=self.password, allow_agent=False, - look_for_keys=False, port=self.port) +"""Network device connection classes for SSH and Telnet.""" + +import re +import time +from typing import Any + +import paramiko +import telnetlib + +from netlib.models import ( + CommandResponse, + EnableModeResponse, + SSHConnectionConfig, + TelnetConnectionConfig, +) + + +class SSH: + """SSH connection handler for network devices.""" + + def __init__( + self, + device_name: str, + username: str, + password: str, + buffer: int | str = 65535, + delay: int | str | float = 1, + port: int | str = 22, + ) -> None: + """Initialize SSH connection. + + Args: + device_name: Device hostname or IP address + username: Username for authentication + password: Password for authentication + buffer: Buffer size for receiving data (default: 65535) + delay: Delay in seconds between operations (default: 1) + port: SSH port (default: 22) + """ + # Validate inputs using Pydantic + config = SSHConnectionConfig( + device_name=device_name, + username=username, + password=password, + buffer=int(buffer), + delay=float(delay), + port=int(port), + ) + + self.device_name = config.device_name + self.username = config.username + self.password = config.password + self.buffer = config.buffer + self.delay = config.delay + self.port = config.port + + self.pre_conn: paramiko.SSHClient | None = None + self.client_conn: paramiko.Channel | None = None + + def connect(self) -> bytes: + """Establish SSH connection to the device. + + Returns: + Initial output from the device after connection + + Raises: + paramiko.SSHException: If connection fails + """ + self.pre_conn = paramiko.SSHClient() + self.pre_conn.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + self.pre_conn.connect( + self.device_name, + username=self.username, + password=self.password, + allow_agent=False, + look_for_keys=False, + port=self.port, + ) self.client_conn = self.pre_conn.invoke_shell() - self.time.sleep(float(self.delay)) - return self.client_conn.recv(self.buffer) + time.sleep(self.delay) + if self.client_conn: + return self.client_conn.recv(self.buffer) + return b"" - def close(self): - return self.pre_conn.close() + def close(self) -> None: + """Close the SSH connection.""" + if self.pre_conn: + self.pre_conn.close() - def clear_buffer(self): - if self.client_conn.recv_ready(): - return self.client_conn.recv(self.buffer).decode('utf-8', 'ignore') - else: - return None + def clear_buffer(self) -> str | None: + """Clear the receive buffer. + + Returns: + Buffer contents if available, None otherwise + """ + if self.client_conn and self.client_conn.recv_ready(): + return self.client_conn.recv(self.buffer).decode("utf-8", "ignore") + return None + + def set_enable(self, enable_password: str) -> str: + """Enter privileged/enable mode. - def set_enable(self, enable_password): - if self.re.search('>$', self.command('\n')): - enable = self.command('enable') - if self.re.search('Password', enable): + Args: + enable_password: Enable password + + Returns: + Response message from the operation + """ + current_prompt = self.command("\n") + if re.search(r">$", current_prompt): + enable_output = self.command("enable") + if re.search("Password", enable_output): send_pwd = self.command(enable_password) return send_pwd - elif self.re.search('#$', self.command('\n')): + elif re.search(r"#$", current_prompt): return "Action: None. Already in enable mode." else: return "Error: Unable to determine user privilege status." + return "Error: Unknown state" + + def disable_paging(self, command: str = "term len 0") -> None: + """Disable paging on the device. + + Args: + command: Command to disable paging (default: 'term len 0') + """ + if self.client_conn: + self.client_conn.sendall(f"{command}\n") + self.clear_buffer() + + def command(self, command: str) -> str: + """Execute a command on the device. + + Args: + command: Command to execute - def disable_paging(self, command='term len 0'): - self.client_conn.sendall(command + "\n") - self.clear_buffer() + Returns: + Command output + """ + if not self.client_conn: + return "" - def command(self, command): - self.client_conn.sendall(command + "\n") + self.client_conn.sendall(f"{command}\n") not_done = True - output = str() + output = "" while not_done: - self.time.sleep(float(self.delay)) + time.sleep(self.delay) if self.client_conn.recv_ready(): - output += self.client_conn.recv(self.buffer).decode('utf-8') + output += self.client_conn.recv(self.buffer).decode("utf-8") else: not_done = False return output - def commands(self, commands_list): - output = str() - if list(commands_list): - for command in commands_list: - output += self.command(command) + def commands(self, commands_list: list[str] | str) -> str: + """Execute multiple commands. + + Args: + commands_list: List of commands or single command string + + Returns: + Combined output from all commands + """ + output = "" + if isinstance(commands_list, list): + for cmd in commands_list: + output += self.command(cmd) else: output += self.command(commands_list) return output -class Telnet(object): - - def __init__(self, device_name, username, password, delay="2", port="23"): - import telnetlib - import time - import re - self.telnetlib = telnetlib - self.time = time - self.re = re - self.device_name = device_name - self.username = username - self.password = password - self.delay = float(delay) - self.port = int(port) - - def connect(self): - self.access = self.telnetlib.Telnet(self.device_name, self.port) - login_prompt = self.access.read_until(b"\(Username: \)|\(login: \)", - self.delay) - if b'login' in login_prompt: +class Telnet: + """Telnet connection handler for network devices.""" + + def __init__( + self, + device_name: str, + username: str, + password: str, + delay: int | str | float = 2, + port: int | str = 23, + ) -> None: + """Initialize Telnet connection. + + Args: + device_name: Device hostname or IP address + username: Username for authentication + password: Password for authentication + delay: Delay in seconds between operations (default: 2) + port: Telnet port (default: 23) + """ + # Validate inputs using Pydantic + config = TelnetConnectionConfig( + device_name=device_name, + username=username, + password=password, + delay=float(delay), + port=int(port), + ) + + self.device_name = config.device_name + self.username = config.username + self.password = config.password + self.delay = config.delay + self.port = config.port + + self.access: telnetlib.Telnet | None = None + self.is_nexus: bool = False + + def connect(self) -> telnetlib.Telnet: + """Establish Telnet connection to the device. + + Returns: + Telnet connection object + + Raises: + OSError: If connection fails + """ + self.access = telnetlib.Telnet(self.device_name, self.port) + login_prompt = self.access.read_until( + b"(Username: )|(login: )", self.delay + ) + if b"login" in login_prompt: self.is_nexus = True - self.access.write(self.username.encode('ascii') + b'\n') - elif b'Username' in login_prompt: + self.access.write(self.username.encode("ascii") + b"\n") + elif b"Username" in login_prompt: self.is_nexus = False - self.access.write(self.username.encode('ascii') + b'\n') - password_prompt = self.access.read_until(b'Password:', - self.delay) - self.access.write(self.password.encode('ascii') + b'\n') + self.access.write(self.username.encode("ascii") + b"\n") + self.access.read_until(b"Password:", self.delay) + self.access.write(self.password.encode("ascii") + b"\n") return self.access - def close(self): - return self.access.close() + def close(self) -> None: + """Close the Telnet connection.""" + if self.access: + self.access.close() + + def clear_buffer(self) -> None: + """Clear the receive buffer (no-op for Telnet).""" - def clear_buffer(self): - pass + def set_enable(self, enable_password: str) -> str: + """Enter privileged/enable mode. - def set_enable(self, enable_password): - if self.re.search(b'>$', self.command('\n')): - self.access.write(b'enable\n') - enable = self.access.read_until(b'Password') - return self.access.write(enable_password.encode('ascii') + b'\n') - elif self.re.search(b'#$', self.command('\n')): + Args: + enable_password: Enable password + + Returns: + Response message from the operation + """ + if not self.access: + return "Error: Not connected" + + current_prompt = self.command("\n") + if re.search(b">$", current_prompt): + self.access.write(b"enable\n") + self.access.read_until(b"Password") + self.access.write(enable_password.encode("ascii") + b"\n") + return "Entered enable mode" + elif re.search(b"#$", current_prompt): return "Action: None. Already in enable mode." else: return "Error: Unable to determine user privilege status." - def disable_paging(self, command='term len 0'): - self.access.write(command.encode('ascii') + b'\n') - return self.access.read_until(b"\(#\)|\(>\)", self.delay) + def disable_paging(self, command: str = "term len 0") -> bytes: + """Disable paging on the device. + + Args: + command: Command to disable paging (default: 'term len 0') + + Returns: + Command output + """ + if not self.access: + return b"" + self.access.write(command.encode("ascii") + b"\n") + return self.access.read_until(b"(#)|(>)", self.delay) + + def command(self, command: str) -> bytes: + """Execute a command on the device. - def command(self, command): - self.access.write(command.encode('ascii') + b'\n') - return self.access.read_until(b"\(#\)|\(>\)", self.delay) + Args: + command: Command to execute - def commands(self, commands_list): - output = str() - if list(commands_list): - for command in commands_list: - output += self.command(command) + Returns: + Command output + """ + if not self.access: + return b"" + self.access.write(command.encode("ascii") + b"\n") + return self.access.read_until(b"(#)|(>)", self.delay) + + def commands(self, commands_list: list[str] | str) -> str: + """Execute multiple commands. + + Args: + commands_list: List of commands or single command string + + Returns: + Combined output from all commands + """ + output = "" + if isinstance(commands_list, list): + for cmd in commands_list: + output += self.command(cmd).decode("utf-8", "ignore") else: - output += self.command(commands_list) + output += self.command(commands_list).decode("utf-8", "ignore") return output diff --git a/netlib/models.py b/netlib/models.py new file mode 100644 index 0000000..923e338 --- /dev/null +++ b/netlib/models.py @@ -0,0 +1,73 @@ +"""Pydantic models for input and output validation.""" + +from typing import Any + +from pydantic import BaseModel, Field, field_validator + + +class ConnectionConfig(BaseModel): + """Configuration for network device connections.""" + + device_name: str = Field(..., description="Device hostname or IP address", min_length=1) + username: str = Field(..., description="Username for authentication", min_length=1) + password: str = Field(..., description="Password for authentication", min_length=1) + port: int = Field(default=22, description="Connection port", ge=1, le=65535) + buffer: int = Field(default=65535, description="Buffer size for data reception", ge=1024) + delay: float = Field(default=1.0, description="Delay in seconds", ge=0.1, le=60.0) + + model_config = {"frozen": False, "validate_assignment": True} + + +class SSHConnectionConfig(ConnectionConfig): + """SSH-specific connection configuration.""" + + port: int = Field(default=22, description="SSH port", ge=1, le=65535) + delay: float = Field(default=1.0, description="Delay in seconds", ge=0.1, le=60.0) + + +class TelnetConnectionConfig(ConnectionConfig): + """Telnet-specific connection configuration.""" + + port: int = Field(default=23, description="Telnet port", ge=1, le=65535) + delay: float = Field(default=2.0, description="Delay in seconds", ge=0.1, le=60.0) + + +class CommandResponse(BaseModel): + """Response from a device command.""" + + output: str = Field(..., description="Command output") + success: bool = Field(default=True, description="Whether command executed successfully") + error: str | None = Field(default=None, description="Error message if any") + + model_config = {"frozen": False} + + +class CredentialsData(BaseModel): + """User credentials data.""" + + username: str = Field(..., description="Username", min_length=1) + password: str = Field(..., description="User password", min_length=1) + enable: str = Field(..., description="Enable/privileged password", min_length=1) + + model_config = {"frozen": False} + + @field_validator("password", "enable") + @classmethod + def validate_password(cls, v: str) -> str: + """Validate password is not empty.""" + if not v or not v.strip(): + msg = "Password cannot be empty" + raise ValueError(msg) + return v + + +class EnableModeResponse(BaseModel): + """Response from enable mode operations.""" + + message: str = Field(..., description="Response message") + already_enabled: bool = Field( + default=False, description="Whether already in enable mode" + ) + success: bool = Field(default=True, description="Whether operation succeeded") + + model_config = {"frozen": False} diff --git a/netlib/user_keyring.py b/netlib/user_keyring.py index d20da0d..63980ae 100644 --- a/netlib/user_keyring.py +++ b/netlib/user_keyring.py @@ -1,64 +1,103 @@ -class KeyRing(object): +"""User credential management using system keyring.""" - def __init__(self, username): - import keyring - import getpass +import getpass + +import keyring + +from netlib.models import CredentialsData + + +class KeyRing: + """Manage user credentials securely using the system keyring.""" + + def __init__(self, username: str) -> None: + """Initialize KeyRing with a username. + + Args: + username: Username for credential storage + """ self.username = username - self.keyring = keyring - self.getpass = getpass - def get_creds(self): - if self.keyring.get_password('nl_user_pass', - username=self.username) is None: - print('No credentials keyring exist. Creating new credentials.') + def get_creds(self) -> dict[str, str]: + """Retrieve credentials from the keyring. + + If credentials don't exist, prompts user to create them. + + Returns: + Dictionary containing username, password, and enable password + """ + user_pass = keyring.get_password("nl_user_pass", self.username) + if user_pass is None: + print("No credentials keyring exist. Creating new credentials.") self.set_creds() - else: - user_pass = self.keyring.get_password('nl_user_pass', - username=self.username) - enable_pass = self.keyring.get_password('nl_enable_pass', - username=self.username) - return {'username': self.username, - 'password': str(user_pass), - 'enable': str(enable_pass)} - - def set_creds(self): + user_pass = keyring.get_password("nl_user_pass", self.username) + + enable_pass = keyring.get_password("nl_enable_pass", self.username) + + # Validate with Pydantic + creds = CredentialsData( + username=self.username, + password=str(user_pass or ""), + enable=str(enable_pass or ""), + ) + + return { + "username": creds.username, + "password": creds.password, + "enable": creds.enable, + } + + def set_creds(self) -> None: + """Set or update credentials in the keyring. + + Prompts user for password and enable password with confirmation. + """ + # Get user password with confirmation match = False - while match is False: - password1 = self.getpass.getpass('Enter your user password: ') - password2 = self.getpass.getpass('Confirm your user password: ') + while not match: + password1 = getpass.getpass("Enter your user password: ") + password2 = getpass.getpass("Confirm your user password: ") if password1 == password2: user_password = password1 match = True + + # Get enable password with confirmation match = False - while match is False: - password1 = self.getpass.getpass('Enter your enable password: ') - password2 = self.getpass.getpass('Confirm your enable password: ') + while not match: + password1 = getpass.getpass("Enter your enable password: ") + password2 = getpass.getpass("Confirm your enable password: ") if password1 == password2: enable_password = password1 match = True - user_pass = self.keyring.set_password('nl_user_pass', - username=self.username, - password=user_password) - enable_pass = self.keyring.set_password('nl_enable_pass', - username=self.username, - password=enable_password) + + # Validate with Pydantic before storing + creds = CredentialsData( + username=self.username, + password=user_password, + enable=enable_password, + ) + + # Store in keyring + keyring.set_password("nl_user_pass", self.username, creds.password) + keyring.set_password("nl_enable_pass", self.username, creds.enable) + self.get_creds() - def del_creds(self): + def del_creds(self) -> None: + """Delete credentials from the keyring. + + Requires password confirmation. Maximum 5 attempts. + """ tries = 0 max_tries = 5 while tries <= max_tries: - user_pass = self.keyring.get_password('nl_user_pass', - username=self.username) - ask_pass = self.getpass.getpass('Enter your user password: ') + user_pass = keyring.get_password("nl_user_pass", self.username) + ask_pass = getpass.getpass("Enter your user password: ") if user_pass == ask_pass: - print('Deleting keyring credentials for {}'.format( - self.username)) - self.keyring.delete_password('nl_user_pass', - username=self.username) - self.keyring.delete_password('nl_enable_pass', - username=self.username) + print(f"Deleting keyring credentials for {self.username}") + keyring.delete_password("nl_user_pass", self.username) + keyring.delete_password("nl_enable_pass", self.username) tries = max_tries + 1 else: tries += 1 - print('Error: Incorrect password.') + print("Error: Incorrect password.") diff --git a/py.typed b/py.typed new file mode 100644 index 0000000..cb5707d --- /dev/null +++ b/py.typed @@ -0,0 +1,2 @@ +# Marker file for PEP 561 +# This package supports type checking diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f38bdbf --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,142 @@ +[tool.poetry] +name = "netlib" +version = "0.2.0" +description = "Simple access to network devices, such as routers and switches, via Telnet and SSH." +authors = ["James Williams"] +license = "MIT" +readme = "README.md" +homepage = "https://github.com/netopsio/netlib" +repository = "https://github.com/netopsio/netlib" +keywords = ["network", "ssh", "telnet", "cisco", "router", "switch"] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "Intended Audience :: System Administrators", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: System :: Networking", +] +packages = [{include = "netlib"}] + +[tool.poetry.dependencies] +python = "^3.10" +paramiko = "^3.5.0" +keyring = "^25.5.0" +keyrings-alt = "^5.0.2" +pydantic = "^2.10.5" +pydantic-settings = "^2.7.1" + +[tool.poetry.group.dev.dependencies] +pytest = "^8.3.4" +pytest-cov = "^6.0.0" +pytest-mock = "^3.14.0" +ruff = "^0.9.1" +mypy = "^1.14.1" +pylint = "^3.3.4" +types-paramiko = "^3.5.0" + +[build-system] +requires = ["poetry-core"] +build-backend = "poetry.core.masonry.api" + +[tool.ruff] +line-length = 100 +target-version = "py310" + +[tool.ruff.lint] +select = [ + "E", # pycodestyle errors + "W", # pycodestyle warnings + "F", # pyflakes + "I", # isort + "N", # pep8-naming + "UP", # pyupgrade + "B", # flake8-bugbear + "C4", # flake8-comprehensions + "DTZ", # flake8-datetimez + "T10", # flake8-debugger + "EM", # flake8-errmsg + "ISC", # flake8-implicit-str-concat + "ICN", # flake8-import-conventions + "PIE", # flake8-pie + "PT", # flake8-pytest-style + "Q", # flake8-quotes + "RSE", # flake8-raise + "RET", # flake8-return + "SIM", # flake8-simplify + "TID", # flake8-tidy-imports + "ARG", # flake8-unused-arguments + "PTH", # flake8-use-pathlib + "PL", # pylint + "RUF", # ruff-specific rules +] +ignore = [ + "PLR0913", # Too many arguments + "PLR2004", # Magic value used in comparison +] + +[tool.ruff.lint.per-file-ignores] +"tests/*" = ["ARG", "PLR2004"] + +[tool.mypy] +python_version = "3.10" +warn_return_any = true +warn_unused_configs = true +disallow_untyped_defs = true +disallow_incomplete_defs = true +check_untyped_defs = true +disallow_untyped_decorators = false +no_implicit_optional = true +warn_redundant_casts = true +warn_unused_ignores = true +warn_no_return = true +strict_equality = true +strict_concatenate = true + +[[tool.mypy.overrides]] +module = [ + "keyring.*", +] +ignore_missing_imports = true + +[tool.pylint.main] +py-version = "3.10" +jobs = 0 +disable = [ + "too-few-public-methods", + "too-many-arguments", + "too-many-instance-attributes", + "missing-module-docstring", + "missing-class-docstring", + "missing-function-docstring", + "line-too-long", # handled by ruff + "invalid-name", +] + +[tool.pytest.ini_options] +minversion = "8.0" +testpaths = ["tests"] +python_files = ["test_*.py"] +python_classes = ["Test*"] +python_functions = ["test_*"] +addopts = [ + "--verbose", + "--strict-markers", + "--cov=netlib", + "--cov-report=term-missing", + "--cov-report=html", + "--cov-report=xml", +] + +[tool.coverage.run] +source = ["netlib"] +omit = ["tests/*", "*/site-packages/*"] + +[tool.coverage.report] +precision = 2 +show_missing = true +skip_covered = false diff --git a/requirements.txt b/requirements.txt deleted file mode 100644 index 07e7eed..0000000 --- a/requirements.txt +++ /dev/null @@ -1,4 +0,0 @@ -keyring>=9.3 -keyrings.alt>=1.1.1 -paramiko>=2.0.1 -pycrypto>=2.6.1 diff --git a/scripts/tests.py b/scripts/tests.py deleted file mode 100755 index 55def12..0000000 --- a/scripts/tests.py +++ /dev/null @@ -1,36 +0,0 @@ -#!/usr/bin/python - -import unittest - -class TestImports(unittest.TestCase): - - def test_import_keyring(self): - try: - from netlib.user_keyring import KeyRing - user_keyring = True - except: - user_keyring = False - raise - self.assertTrue(user_keyring) - - def test_import_ssh(self): - try: - from netlib.conn_type import SSH - ssh = True - except: - ssh = False - raise - self.assertTrue(ssh) - - def test_import_telnet(self): - try: - from netlib.conn_type import Telnet - telnet = True - except: - telnet = False - raise - self.assertTrue(telnet) - - -if __name__ == '__main__': - unittest.main() diff --git a/setup.py b/setup.py deleted file mode 100644 index 88d3031..0000000 --- a/setup.py +++ /dev/null @@ -1,23 +0,0 @@ -#!/usr/bin/env python - -from setuptools import setup -from netlib import __version__ - - -setup( - name='netlib', - version=__version__, - url='https://github.com/netopsio/netlib', - author='James Williams', - license='MIT', - install_requires=[ - 'paramiko', - 'pycrypto', - 'keyring', - 'keyrings.alt' - ], - description='Simple access to network devices, such as routers and switches, via Telnet and SSH.', - packages=[ - 'netlib', - ], -) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..53feaad --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for netlib package.""" diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..d11bd89 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,20 @@ +"""Pytest configuration and fixtures.""" + +import pytest + + +@pytest.fixture +def mock_device_config() -> dict[str, str | int]: + """Provide mock device configuration.""" + return { + "device_name": "test-router", + "username": "testuser", + "password": "testpass", + "port": 22, + } + + +@pytest.fixture +def mock_enable_password() -> str: + """Provide mock enable password.""" + return "enablepass" diff --git a/tests/test_imports.py b/tests/test_imports.py new file mode 100644 index 0000000..b366958 --- /dev/null +++ b/tests/test_imports.py @@ -0,0 +1,49 @@ +"""Tests for module imports.""" + +import pytest + + +class TestImports: + """Tests for importing netlib modules.""" + + def test_import_keyring(self) -> None: + """Test importing KeyRing class.""" + from netlib.user_keyring import KeyRing + + assert KeyRing is not None + + def test_import_ssh(self) -> None: + """Test importing SSH class.""" + from netlib.conn_type import SSH + + assert SSH is not None + + def test_import_telnet(self) -> None: + """Test importing Telnet class.""" + from netlib.conn_type import Telnet + + assert Telnet is not None + + def test_import_from_package(self) -> None: + """Test importing from main package.""" + from netlib import SSH, Telnet, KeyRing + + assert SSH is not None + assert Telnet is not None + assert KeyRing is not None + + def test_import_models(self) -> None: + """Test importing Pydantic models.""" + from netlib.models import ( + SSHConnectionConfig, + TelnetConnectionConfig, + CommandResponse, + CredentialsData, + EnableModeResponse, + ) + + assert SSHConnectionConfig is not None + assert TelnetConnectionConfig is not None + assert CommandResponse is not None + assert CredentialsData is not None + assert EnableModeResponse is not None diff --git a/tests/test_keyring.py b/tests/test_keyring.py new file mode 100644 index 0000000..ecbe47d --- /dev/null +++ b/tests/test_keyring.py @@ -0,0 +1,178 @@ +"""Tests for KeyRing credential management.""" + +from unittest.mock import MagicMock, Mock, patch + +import pytest + +from netlib.user_keyring import KeyRing + + +class TestKeyRingInit: + """Tests for KeyRing initialization.""" + + def test_init(self) -> None: + """Test KeyRing initialization.""" + kr = KeyRing("testuser") + assert kr.username == "testuser" + + +class TestKeyRingGetCreds: + """Tests for retrieving credentials.""" + + @patch("netlib.user_keyring.keyring.get_password") + def test_get_existing_credentials(self, mock_get_password: Mock) -> None: + """Test retrieving existing credentials.""" + mock_get_password.side_effect = ["userpass123", "enablepass123"] + + kr = KeyRing("testuser") + creds = kr.get_creds() + + assert creds["username"] == "testuser" + assert creds["password"] == "userpass123" + assert creds["enable"] == "enablepass123" + + @patch("netlib.user_keyring.keyring.get_password") + @patch("netlib.user_keyring.getpass.getpass") + @patch("netlib.user_keyring.keyring.set_password") + def test_get_creds_creates_new_when_none_exist( + self, + mock_set_password: Mock, + mock_getpass: Mock, + mock_get_password: Mock, + ) -> None: + """Test creating new credentials when none exist.""" + # First call returns None (no creds), subsequent calls return passwords + mock_get_password.side_effect = [ + None, # First check - no password exists + "newuserpass", # After setting + "newenablepass", # After setting + ] + mock_getpass.side_effect = [ + "newuserpass", # Enter user password + "newuserpass", # Confirm user password + "newenablepass", # Enter enable password + "newenablepass", # Confirm enable password + ] + + kr = KeyRing("testuser") + with patch("builtins.print"): + creds = kr.get_creds() + + assert creds["username"] == "testuser" + assert creds["password"] == "newuserpass" + assert creds["enable"] == "newenablepass" + + +class TestKeyRingSetCreds: + """Tests for setting credentials.""" + + @patch("netlib.user_keyring.keyring.set_password") + @patch("netlib.user_keyring.keyring.get_password") + @patch("netlib.user_keyring.getpass.getpass") + def test_set_credentials_matching_passwords( + self, + mock_getpass: Mock, + mock_get_password: Mock, + mock_set_password: Mock, + ) -> None: + """Test setting credentials with matching passwords.""" + mock_getpass.side_effect = [ + "password1", # Enter user password + "password1", # Confirm user password + "enablepass1", # Enter enable password + "enablepass1", # Confirm enable password + ] + mock_get_password.side_effect = ["password1", "enablepass1"] + + kr = KeyRing("testuser") + kr.set_creds() + + assert mock_set_password.call_count == 2 + + @patch("netlib.user_keyring.keyring.set_password") + @patch("netlib.user_keyring.keyring.get_password") + @patch("netlib.user_keyring.getpass.getpass") + def test_set_credentials_retry_on_mismatch( + self, + mock_getpass: Mock, + mock_get_password: Mock, + mock_set_password: Mock, + ) -> None: + """Test password retry when confirmation doesn't match.""" + mock_getpass.side_effect = [ + "password1", # Enter user password + "wrongpass", # Confirm user password (mismatch) + "password1", # Enter user password (retry) + "password1", # Confirm user password (match) + "enablepass1", # Enter enable password + "enablepass1", # Confirm enable password + ] + mock_get_password.side_effect = ["password1", "enablepass1"] + + kr = KeyRing("testuser") + kr.set_creds() + + assert mock_set_password.call_count == 2 + + +class TestKeyRingDelCreds: + """Tests for deleting credentials.""" + + @patch("netlib.user_keyring.keyring.delete_password") + @patch("netlib.user_keyring.keyring.get_password") + @patch("netlib.user_keyring.getpass.getpass") + def test_delete_credentials_correct_password( + self, + mock_getpass: Mock, + mock_get_password: Mock, + mock_delete_password: Mock, + ) -> None: + """Test deleting credentials with correct password.""" + mock_get_password.return_value = "correctpass" + mock_getpass.return_value = "correctpass" + + kr = KeyRing("testuser") + with patch("builtins.print"): + kr.del_creds() + + assert mock_delete_password.call_count == 2 + + @patch("netlib.user_keyring.keyring.delete_password") + @patch("netlib.user_keyring.keyring.get_password") + @patch("netlib.user_keyring.getpass.getpass") + def test_delete_credentials_incorrect_password( + self, + mock_getpass: Mock, + mock_get_password: Mock, + mock_delete_password: Mock, + ) -> None: + """Test attempting to delete with incorrect password.""" + mock_get_password.return_value = "correctpass" + mock_getpass.side_effect = ["wrongpass"] * 6 # Will fail all attempts + + kr = KeyRing("testuser") + with patch("builtins.print"): + kr.del_creds() + + # Should not delete anything + mock_delete_password.assert_not_called() + + @patch("netlib.user_keyring.keyring.delete_password") + @patch("netlib.user_keyring.keyring.get_password") + @patch("netlib.user_keyring.getpass.getpass") + def test_delete_credentials_retry_on_wrong_password( + self, + mock_getpass: Mock, + mock_get_password: Mock, + mock_delete_password: Mock, + ) -> None: + """Test retry mechanism when wrong password is entered.""" + mock_get_password.return_value = "correctpass" + mock_getpass.side_effect = ["wrongpass", "correctpass"] + + kr = KeyRing("testuser") + with patch("builtins.print"): + kr.del_creds() + + # Should delete after second attempt + assert mock_delete_password.call_count == 2 diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..5cc93c7 --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,157 @@ +"""Tests for Pydantic models.""" + +import pytest +from pydantic import ValidationError + +from netlib.models import ( + CommandResponse, + CredentialsData, + EnableModeResponse, + SSHConnectionConfig, + TelnetConnectionConfig, +) + + +class TestSSHConnectionConfig: + """Tests for SSH connection configuration.""" + + def test_valid_config(self) -> None: + """Test valid SSH configuration.""" + config = SSHConnectionConfig( + device_name="router1", + username="admin", + password="secret", + ) + assert config.device_name == "router1" + assert config.username == "admin" + assert config.password == "secret" + assert config.port == 22 + assert config.buffer == 65535 + assert config.delay == 1.0 + + def test_custom_port(self) -> None: + """Test custom SSH port.""" + config = SSHConnectionConfig( + device_name="router1", + username="admin", + password="secret", + port=2222, + ) + assert config.port == 2222 + + def test_invalid_port(self) -> None: + """Test invalid port number.""" + with pytest.raises(ValidationError): + SSHConnectionConfig( + device_name="router1", + username="admin", + password="secret", + port=70000, + ) + + def test_empty_device_name(self) -> None: + """Test empty device name.""" + with pytest.raises(ValidationError): + SSHConnectionConfig( + device_name="", + username="admin", + password="secret", + ) + + +class TestTelnetConnectionConfig: + """Tests for Telnet connection configuration.""" + + def test_valid_config(self) -> None: + """Test valid Telnet configuration.""" + config = TelnetConnectionConfig( + device_name="switch1", + username="admin", + password="secret", + ) + assert config.device_name == "switch1" + assert config.port == 23 + assert config.delay == 2.0 + + def test_custom_delay(self) -> None: + """Test custom delay value.""" + config = TelnetConnectionConfig( + device_name="switch1", + username="admin", + password="secret", + delay=5.0, + ) + assert config.delay == 5.0 + + +class TestCommandResponse: + """Tests for command response model.""" + + def test_successful_response(self) -> None: + """Test successful command response.""" + response = CommandResponse(output="show version output") + assert response.output == "show version output" + assert response.success is True + assert response.error is None + + def test_error_response(self) -> None: + """Test error command response.""" + response = CommandResponse( + output="", + success=False, + error="Command not found", + ) + assert response.success is False + assert response.error == "Command not found" + + +class TestCredentialsData: + """Tests for credentials data model.""" + + def test_valid_credentials(self) -> None: + """Test valid credentials.""" + creds = CredentialsData( + username="admin", + password="userpass", + enable="enablepass", + ) + assert creds.username == "admin" + assert creds.password == "userpass" + assert creds.enable == "enablepass" + + def test_empty_password(self) -> None: + """Test empty password validation.""" + with pytest.raises(ValidationError): + CredentialsData( + username="admin", + password="", + enable="enablepass", + ) + + def test_whitespace_password(self) -> None: + """Test whitespace-only password validation.""" + with pytest.raises(ValidationError): + CredentialsData( + username="admin", + password=" ", + enable="enablepass", + ) + + +class TestEnableModeResponse: + """Tests for enable mode response model.""" + + def test_successful_enable(self) -> None: + """Test successful enable mode entry.""" + response = EnableModeResponse(message="Entered enable mode") + assert response.message == "Entered enable mode" + assert response.success is True + assert response.already_enabled is False + + def test_already_enabled(self) -> None: + """Test already in enable mode.""" + response = EnableModeResponse( + message="Already in enable mode", + already_enabled=True, + ) + assert response.already_enabled is True diff --git a/tests/test_ssh.py b/tests/test_ssh.py new file mode 100644 index 0000000..d10e8d4 --- /dev/null +++ b/tests/test_ssh.py @@ -0,0 +1,200 @@ +"""Tests for SSH connection class.""" + +from unittest.mock import MagicMock, Mock, patch + +import pytest + +from netlib.conn_type import SSH + + +class TestSSHInit: + """Tests for SSH initialization.""" + + def test_init_with_defaults(self) -> None: + """Test SSH initialization with default values.""" + ssh = SSH("router1", "admin", "secret") + assert ssh.device_name == "router1" + assert ssh.username == "admin" + assert ssh.password == "secret" + assert ssh.port == 22 + assert ssh.buffer == 65535 + assert ssh.delay == 1.0 + + def test_init_with_custom_values(self) -> None: + """Test SSH initialization with custom values.""" + ssh = SSH( + "router1", + "admin", + "secret", + buffer=8192, + delay=2, + port=2222, + ) + assert ssh.port == 2222 + assert ssh.buffer == 8192 + assert ssh.delay == 2.0 + + def test_init_with_string_values(self) -> None: + """Test SSH initialization with string numeric values.""" + ssh = SSH("router1", "admin", "secret", buffer="8192", delay="2", port="2222") + assert ssh.port == 2222 + assert ssh.buffer == 8192 + assert ssh.delay == 2.0 + + +class TestSSHConnect: + """Tests for SSH connect method.""" + + @patch("netlib.conn_type.paramiko.SSHClient") + @patch("netlib.conn_type.time.sleep") + def test_connect_success( + self, mock_sleep: Mock, mock_ssh_client: Mock + ) -> None: + """Test successful SSH connection.""" + # Setup mocks + mock_client = MagicMock() + mock_ssh_client.return_value = mock_client + mock_channel = MagicMock() + mock_client.invoke_shell.return_value = mock_channel + mock_channel.recv.return_value = b"Router> " + + # Create SSH instance and connect + ssh = SSH("router1", "admin", "secret") + result = ssh.connect() + + # Assertions + mock_client.set_missing_host_key_policy.assert_called_once() + mock_client.connect.assert_called_once_with( + "router1", + username="admin", + password="secret", + allow_agent=False, + look_for_keys=False, + port=22, + ) + mock_client.invoke_shell.assert_called_once() + assert result == b"Router> " + + +class TestSSHCommand: + """Tests for SSH command execution.""" + + def test_command_no_connection(self) -> None: + """Test command execution without connection.""" + ssh = SSH("router1", "admin", "secret") + result = ssh.command("show version") + assert result == "" + + @patch("netlib.conn_type.time.sleep") + def test_command_with_connection(self, mock_sleep: Mock) -> None: + """Test command execution with active connection.""" + ssh = SSH("router1", "admin", "secret") + + # Mock the client connection + mock_channel = MagicMock() + ssh.client_conn = mock_channel + + # Setup recv_ready to return True once, then False + mock_channel.recv_ready.side_effect = [True, False] + mock_channel.recv.return_value = b"Version output" + + result = ssh.command("show version") + + mock_channel.sendall.assert_called_once_with("show version\n") + assert "Version output" in result + + +class TestSSHCommands: + """Tests for SSH multiple commands execution.""" + + def test_commands_with_list(self) -> None: + """Test executing multiple commands from list.""" + ssh = SSH("router1", "admin", "secret") + + with patch.object(ssh, "command", return_value="output\n") as mock_cmd: + result = ssh.commands(["show version", "show interfaces"]) + + assert mock_cmd.call_count == 2 + assert "output" in result + + def test_commands_with_string(self) -> None: + """Test executing single command as string.""" + ssh = SSH("router1", "admin", "secret") + + with patch.object(ssh, "command", return_value="output\n") as mock_cmd: + result = ssh.commands("show version") + + mock_cmd.assert_called_once_with("show version") + assert "output" in result + + +class TestSSHClearBuffer: + """Tests for SSH buffer clearing.""" + + def test_clear_buffer_no_data(self) -> None: + """Test clearing buffer when no data available.""" + ssh = SSH("router1", "admin", "secret") + + mock_channel = MagicMock() + mock_channel.recv_ready.return_value = False + ssh.client_conn = mock_channel + + result = ssh.clear_buffer() + assert result is None + + def test_clear_buffer_with_data(self) -> None: + """Test clearing buffer with data available.""" + ssh = SSH("router1", "admin", "secret") + + mock_channel = MagicMock() + mock_channel.recv_ready.return_value = True + mock_channel.recv.return_value = b"Buffer data" + ssh.client_conn = mock_channel + + result = ssh.clear_buffer() + assert result == "Buffer data" + + +class TestSSHDisablePaging: + """Tests for disabling paging.""" + + def test_disable_paging(self) -> None: + """Test disable paging command.""" + ssh = SSH("router1", "admin", "secret") + + mock_channel = MagicMock() + ssh.client_conn = mock_channel + + with patch.object(ssh, "clear_buffer"): + ssh.disable_paging() + mock_channel.sendall.assert_called_once_with("term len 0\n") + + def test_disable_paging_custom_command(self) -> None: + """Test disable paging with custom command.""" + ssh = SSH("router1", "admin", "secret") + + mock_channel = MagicMock() + ssh.client_conn = mock_channel + + with patch.object(ssh, "clear_buffer"): + ssh.disable_paging("terminal length 0") + mock_channel.sendall.assert_called_once_with("terminal length 0\n") + + +class TestSSHClose: + """Tests for SSH connection closing.""" + + def test_close_connection(self) -> None: + """Test closing SSH connection.""" + ssh = SSH("router1", "admin", "secret") + + mock_client = MagicMock() + ssh.pre_conn = mock_client + + ssh.close() + mock_client.close.assert_called_once() + + def test_close_no_connection(self) -> None: + """Test closing when no connection exists.""" + ssh = SSH("router1", "admin", "secret") + ssh.close() # Should not raise any exception diff --git a/tests/test_telnet.py b/tests/test_telnet.py new file mode 100644 index 0000000..8fb4075 --- /dev/null +++ b/tests/test_telnet.py @@ -0,0 +1,172 @@ +"""Tests for Telnet connection class.""" + +from unittest.mock import MagicMock, Mock, patch + +import pytest + +from netlib.conn_type import Telnet + + +class TestTelnetInit: + """Tests for Telnet initialization.""" + + def test_init_with_defaults(self) -> None: + """Test Telnet initialization with default values.""" + telnet = Telnet("switch1", "admin", "secret") + assert telnet.device_name == "switch1" + assert telnet.username == "admin" + assert telnet.password == "secret" + assert telnet.port == 23 + assert telnet.delay == 2.0 + + def test_init_with_custom_values(self) -> None: + """Test Telnet initialization with custom values.""" + telnet = Telnet("switch1", "admin", "secret", delay=5, port=2323) + assert telnet.port == 2323 + assert telnet.delay == 5.0 + + def test_init_with_string_values(self) -> None: + """Test Telnet initialization with string numeric values.""" + telnet = Telnet("switch1", "admin", "secret", delay="3", port="2323") + assert telnet.port == 2323 + assert telnet.delay == 3.0 + + +class TestTelnetConnect: + """Tests for Telnet connect method.""" + + @patch("netlib.conn_type.telnetlib.Telnet") + def test_connect_with_username_prompt(self, mock_telnet_class: Mock) -> None: + """Test Telnet connection with Username prompt.""" + mock_telnet = MagicMock() + mock_telnet_class.return_value = mock_telnet + mock_telnet.read_until.side_effect = [ + b"Username: ", + b"Password:", + ] + + telnet = Telnet("switch1", "admin", "secret") + result = telnet.connect() + + mock_telnet_class.assert_called_once_with("switch1", 23) + assert telnet.is_nexus is False + assert result == mock_telnet + + @patch("netlib.conn_type.telnetlib.Telnet") + def test_connect_with_login_prompt(self, mock_telnet_class: Mock) -> None: + """Test Telnet connection with login prompt (Nexus).""" + mock_telnet = MagicMock() + mock_telnet_class.return_value = mock_telnet + mock_telnet.read_until.side_effect = [ + b"login: ", + b"Password:", + ] + + telnet = Telnet("switch1", "admin", "secret") + result = telnet.connect() + + assert telnet.is_nexus is True + assert result == mock_telnet + + +class TestTelnetCommand: + """Tests for Telnet command execution.""" + + def test_command_no_connection(self) -> None: + """Test command execution without connection.""" + telnet = Telnet("switch1", "admin", "secret") + result = telnet.command("show version") + assert result == b"" + + def test_command_with_connection(self) -> None: + """Test command execution with active connection.""" + telnet = Telnet("switch1", "admin", "secret") + + mock_connection = MagicMock() + mock_connection.read_until.return_value = b"Version output\nSwitch#" + telnet.access = mock_connection + + result = telnet.command("show version") + + mock_connection.write.assert_called_once_with(b"show version\n") + assert b"Version output" in result + + +class TestTelnetCommands: + """Tests for Telnet multiple commands execution.""" + + def test_commands_with_list(self) -> None: + """Test executing multiple commands from list.""" + telnet = Telnet("switch1", "admin", "secret") + + with patch.object( + telnet, "command", return_value=b"output\n" + ) as mock_cmd: + result = telnet.commands(["show version", "show interfaces"]) + + assert mock_cmd.call_count == 2 + assert "output" in result + + def test_commands_with_string(self) -> None: + """Test executing single command as string.""" + telnet = Telnet("switch1", "admin", "secret") + + with patch.object( + telnet, "command", return_value=b"output\n" + ) as mock_cmd: + result = telnet.commands("show version") + + mock_cmd.assert_called_once_with("show version") + assert "output" in result + + +class TestTelnetDisablePaging: + """Tests for disabling paging.""" + + def test_disable_paging_no_connection(self) -> None: + """Test disable paging without connection.""" + telnet = Telnet("switch1", "admin", "secret") + result = telnet.disable_paging() + assert result == b"" + + def test_disable_paging(self) -> None: + """Test disable paging command.""" + telnet = Telnet("switch1", "admin", "secret") + + mock_connection = MagicMock() + mock_connection.read_until.return_value = b"Switch#" + telnet.access = mock_connection + + result = telnet.disable_paging() + + mock_connection.write.assert_called_once_with(b"term len 0\n") + assert result == b"Switch#" + + +class TestTelnetClearBuffer: + """Tests for Telnet buffer clearing.""" + + def test_clear_buffer(self) -> None: + """Test clear buffer is a no-op for Telnet.""" + telnet = Telnet("switch1", "admin", "secret") + result = telnet.clear_buffer() + assert result is None + + +class TestTelnetClose: + """Tests for Telnet connection closing.""" + + def test_close_connection(self) -> None: + """Test closing Telnet connection.""" + telnet = Telnet("switch1", "admin", "secret") + + mock_connection = MagicMock() + telnet.access = mock_connection + + telnet.close() + mock_connection.close.assert_called_once() + + def test_close_no_connection(self) -> None: + """Test closing when no connection exists.""" + telnet = Telnet("switch1", "admin", "secret") + telnet.close() # Should not raise any exception From e621febcd647567f14d9c674eec268665fbaec86 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 29 Jan 2026 02:29:20 +0000 Subject: [PATCH 2/5] Add SecretStr for password security and dynamic version management ## Security Enhancement - SecretStr for Passwords - Update all password fields to use Pydantic SecretStr type - Models updated: - ConnectionConfig.password now SecretStr - CredentialsData.password and .enable now SecretStr - Classes updated to handle SecretStr: - SSH class stores password as _password (SecretStr) - Telnet class stores password as _password (SecretStr) - Both classes use .get_secret_value() when using passwords - set_enable() methods accept str | SecretStr for flexibility - KeyRing.get_creds() returns plain strings (for backward compatibility) - Internal validation ensures passwords aren't empty - Benefits: - Passwords not exposed in logs or repr() - Better security for credential handling - IDE autocomplete won't leak password values ## Dynamic Version Management - __version__ now reads from pyproject.toml - Uses importlib.metadata when package is installed - Falls back to reading pyproject.toml during development - Added tomli dependency for Python <3.11 (Python 3.11+ uses built-in tomllib) - Graceful fallback to "0.2.0-dev" if neither method works ## Test Updates - Updated all tests to use .get_secret_value() for password assertions - Tests now properly validate SecretStr behavior - Maintained backward compatibility in test patterns https://claude.ai/code/session_019D1TUrg1xPGse1HsS8pgqQ --- netlib/__init__.py | 32 +++++++++++++++++++++++++++++++- netlib/conn_type.py | 35 +++++++++++++++++++++++++---------- netlib/models.py | 23 +++++++++++++++++------ netlib/user_keyring.py | 8 ++++---- pyproject.toml | 1 + tests/test_models.py | 6 +++--- tests/test_ssh.py | 2 +- tests/test_telnet.py | 2 +- 8 files changed, 83 insertions(+), 26 deletions(-) diff --git a/netlib/__init__.py b/netlib/__init__.py index 732a353..1bd9f24 100644 --- a/netlib/__init__.py +++ b/netlib/__init__.py @@ -1,7 +1,37 @@ """NetLib - Network device connection library.""" +import sys +from importlib.metadata import PackageNotFoundError, version +from pathlib import Path + __name__ = "netlib" -__version__ = "0.2.0" + +# Try to get version from installed package metadata +try: + __version__ = version("netlib") +except PackageNotFoundError: + # Fallback: read from pyproject.toml during development + try: + if sys.version_info >= (3, 11): + import tomllib + else: + try: + import tomli as tomllib # type: ignore[import-not-found,no-redef] + except ImportError: + # If tomli not available, use fallback version + __version__ = "0.2.0-dev" + tomllib = None # type: ignore[assignment] + + if tomllib is not None: + pyproject_path = Path(__file__).parent.parent / "pyproject.toml" + if pyproject_path.exists(): + with pyproject_path.open("rb") as f: + pyproject_data = tomllib.load(f) + __version__ = pyproject_data["tool"]["poetry"]["version"] + else: + __version__ = "0.2.0-dev" + except Exception: + __version__ = "0.2.0-dev" from netlib.conn_type import SSH, Telnet from netlib.user_keyring import KeyRing diff --git a/netlib/conn_type.py b/netlib/conn_type.py index 2509600..70915e0 100644 --- a/netlib/conn_type.py +++ b/netlib/conn_type.py @@ -6,6 +6,7 @@ import paramiko import telnetlib +from pydantic import SecretStr from netlib.models import ( CommandResponse, @@ -49,7 +50,7 @@ def __init__( self.device_name = config.device_name self.username = config.username - self.password = config.password + self._password = config.password # SecretStr self.buffer = config.buffer self.delay = config.delay self.port = config.port @@ -71,7 +72,7 @@ def connect(self) -> bytes: self.pre_conn.connect( self.device_name, username=self.username, - password=self.password, + password=self._password.get_secret_value(), allow_agent=False, look_for_keys=False, port=self.port, @@ -97,20 +98,27 @@ def clear_buffer(self) -> str | None: return self.client_conn.recv(self.buffer).decode("utf-8", "ignore") return None - def set_enable(self, enable_password: str) -> str: + def set_enable(self, enable_password: str | SecretStr) -> str: """Enter privileged/enable mode. Args: - enable_password: Enable password + enable_password: Enable password (can be str or SecretStr) Returns: Response message from the operation """ + # Handle both str and SecretStr + pwd = ( + enable_password.get_secret_value() + if isinstance(enable_password, SecretStr) + else enable_password + ) + current_prompt = self.command("\n") if re.search(r">$", current_prompt): enable_output = self.command("enable") if re.search("Password", enable_output): - send_pwd = self.command(enable_password) + send_pwd = self.command(pwd) return send_pwd elif re.search(r"#$", current_prompt): return "Action: None. Already in enable mode." @@ -200,7 +208,7 @@ def __init__( self.device_name = config.device_name self.username = config.username - self.password = config.password + self._password = config.password # SecretStr self.delay = config.delay self.port = config.port @@ -227,7 +235,7 @@ def connect(self) -> telnetlib.Telnet: self.is_nexus = False self.access.write(self.username.encode("ascii") + b"\n") self.access.read_until(b"Password:", self.delay) - self.access.write(self.password.encode("ascii") + b"\n") + self.access.write(self._password.get_secret_value().encode("ascii") + b"\n") return self.access def close(self) -> None: @@ -238,11 +246,11 @@ def close(self) -> None: def clear_buffer(self) -> None: """Clear the receive buffer (no-op for Telnet).""" - def set_enable(self, enable_password: str) -> str: + def set_enable(self, enable_password: str | SecretStr) -> str: """Enter privileged/enable mode. Args: - enable_password: Enable password + enable_password: Enable password (can be str or SecretStr) Returns: Response message from the operation @@ -250,11 +258,18 @@ def set_enable(self, enable_password: str) -> str: if not self.access: return "Error: Not connected" + # Handle both str and SecretStr + pwd = ( + enable_password.get_secret_value() + if isinstance(enable_password, SecretStr) + else enable_password + ) + current_prompt = self.command("\n") if re.search(b">$", current_prompt): self.access.write(b"enable\n") self.access.read_until(b"Password") - self.access.write(enable_password.encode("ascii") + b"\n") + self.access.write(pwd.encode("ascii") + b"\n") return "Entered enable mode" elif re.search(b"#$", current_prompt): return "Action: None. Already in enable mode." diff --git a/netlib/models.py b/netlib/models.py index 923e338..9ed0e7e 100644 --- a/netlib/models.py +++ b/netlib/models.py @@ -2,7 +2,7 @@ from typing import Any -from pydantic import BaseModel, Field, field_validator +from pydantic import BaseModel, Field, SecretStr, field_validator class ConnectionConfig(BaseModel): @@ -10,13 +10,23 @@ class ConnectionConfig(BaseModel): device_name: str = Field(..., description="Device hostname or IP address", min_length=1) username: str = Field(..., description="Username for authentication", min_length=1) - password: str = Field(..., description="Password for authentication", min_length=1) + password: SecretStr = Field(..., description="Password for authentication") port: int = Field(default=22, description="Connection port", ge=1, le=65535) buffer: int = Field(default=65535, description="Buffer size for data reception", ge=1024) delay: float = Field(default=1.0, description="Delay in seconds", ge=0.1, le=60.0) model_config = {"frozen": False, "validate_assignment": True} + @field_validator("password") + @classmethod + def validate_password(cls, v: SecretStr) -> SecretStr: + """Validate password is not empty.""" + secret_value = v.get_secret_value() + if not secret_value or not secret_value.strip(): + msg = "Password cannot be empty" + raise ValueError(msg) + return v + class SSHConnectionConfig(ConnectionConfig): """SSH-specific connection configuration.""" @@ -46,16 +56,17 @@ class CredentialsData(BaseModel): """User credentials data.""" username: str = Field(..., description="Username", min_length=1) - password: str = Field(..., description="User password", min_length=1) - enable: str = Field(..., description="Enable/privileged password", min_length=1) + password: SecretStr = Field(..., description="User password") + enable: SecretStr = Field(..., description="Enable/privileged password") model_config = {"frozen": False} @field_validator("password", "enable") @classmethod - def validate_password(cls, v: str) -> str: + def validate_password_field(cls, v: SecretStr) -> SecretStr: """Validate password is not empty.""" - if not v or not v.strip(): + secret_value = v.get_secret_value() + if not secret_value or not secret_value.strip(): msg = "Password cannot be empty" raise ValueError(msg) return v diff --git a/netlib/user_keyring.py b/netlib/user_keyring.py index 63980ae..ee03501 100644 --- a/netlib/user_keyring.py +++ b/netlib/user_keyring.py @@ -43,8 +43,8 @@ def get_creds(self) -> dict[str, str]: return { "username": creds.username, - "password": creds.password, - "enable": creds.enable, + "password": creds.password.get_secret_value(), + "enable": creds.enable.get_secret_value(), } def set_creds(self) -> None: @@ -78,8 +78,8 @@ def set_creds(self) -> None: ) # Store in keyring - keyring.set_password("nl_user_pass", self.username, creds.password) - keyring.set_password("nl_enable_pass", self.username, creds.enable) + keyring.set_password("nl_user_pass", self.username, creds.password.get_secret_value()) + keyring.set_password("nl_enable_pass", self.username, creds.enable.get_secret_value()) self.get_creds() diff --git a/pyproject.toml b/pyproject.toml index f38bdbf..21570e4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,6 +29,7 @@ keyring = "^25.5.0" keyrings-alt = "^5.0.2" pydantic = "^2.10.5" pydantic-settings = "^2.7.1" +tomli = {version = "^2.2.1", python = "<3.11"} [tool.poetry.group.dev.dependencies] pytest = "^8.3.4" diff --git a/tests/test_models.py b/tests/test_models.py index 5cc93c7..6a084f7 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -24,7 +24,7 @@ def test_valid_config(self) -> None: ) assert config.device_name == "router1" assert config.username == "admin" - assert config.password == "secret" + assert config.password.get_secret_value() == "secret" assert config.port == 22 assert config.buffer == 65535 assert config.delay == 1.0 @@ -116,8 +116,8 @@ def test_valid_credentials(self) -> None: enable="enablepass", ) assert creds.username == "admin" - assert creds.password == "userpass" - assert creds.enable == "enablepass" + assert creds.password.get_secret_value() == "userpass" + assert creds.enable.get_secret_value() == "enablepass" def test_empty_password(self) -> None: """Test empty password validation.""" diff --git a/tests/test_ssh.py b/tests/test_ssh.py index d10e8d4..7f6875c 100644 --- a/tests/test_ssh.py +++ b/tests/test_ssh.py @@ -15,7 +15,7 @@ def test_init_with_defaults(self) -> None: ssh = SSH("router1", "admin", "secret") assert ssh.device_name == "router1" assert ssh.username == "admin" - assert ssh.password == "secret" + assert ssh._password.get_secret_value() == "secret" assert ssh.port == 22 assert ssh.buffer == 65535 assert ssh.delay == 1.0 diff --git a/tests/test_telnet.py b/tests/test_telnet.py index 8fb4075..a0750d5 100644 --- a/tests/test_telnet.py +++ b/tests/test_telnet.py @@ -15,7 +15,7 @@ def test_init_with_defaults(self) -> None: telnet = Telnet("switch1", "admin", "secret") assert telnet.device_name == "switch1" assert telnet.username == "admin" - assert telnet.password == "secret" + assert telnet._password.get_secret_value() == "secret" assert telnet.port == 23 assert telnet.delay == 2.0 From aaf6473bc84c90fbf0852a4c0bb9ae7ab25e1083 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 29 Jan 2026 02:34:02 +0000 Subject: [PATCH 3/5] Apply ruff formatting and linting fixes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Code Quality Improvements - Run ruff format to apply consistent code formatting - Fix all ruff linting issues automatically - Remove unused imports (typing.Any, pytest, MagicMock) - Fix import sorting across all modules - Sort __all__ exports alphabetically ## Manual Fixes - Remove unnecessary variable assignment before return in SSH.set_enable() - Add noqa comment for intentional function-level imports in test_imports.py - Update pyproject.toml per-file-ignores configuration ## Results - All ruff format checks pass โœ“ - All ruff linting checks pass โœ“ - Code follows consistent style guidelines - Improved code maintainability and readability https://claude.ai/code/session_019D1TUrg1xPGse1HsS8pgqQ --- netlib/__init__.py | 2 +- netlib/conn_type.py | 17 +++++------------ netlib/models.py | 6 +----- pyproject.toml | 1 + tests/test_imports.py | 8 ++++---- tests/test_keyring.py | 4 +--- tests/test_ssh.py | 6 +----- tests/test_telnet.py | 10 ++-------- 8 files changed, 16 insertions(+), 38 deletions(-) diff --git a/netlib/__init__.py b/netlib/__init__.py index 1bd9f24..b852a13 100644 --- a/netlib/__init__.py +++ b/netlib/__init__.py @@ -36,4 +36,4 @@ from netlib.conn_type import SSH, Telnet from netlib.user_keyring import KeyRing -__all__ = ["SSH", "Telnet", "KeyRing"] +__all__ = ["SSH", "KeyRing", "Telnet"] diff --git a/netlib/conn_type.py b/netlib/conn_type.py index 70915e0..36e028b 100644 --- a/netlib/conn_type.py +++ b/netlib/conn_type.py @@ -1,16 +1,13 @@ """Network device connection classes for SSH and Telnet.""" import re +import telnetlib import time -from typing import Any import paramiko -import telnetlib from pydantic import SecretStr from netlib.models import ( - CommandResponse, - EnableModeResponse, SSHConnectionConfig, TelnetConnectionConfig, ) @@ -118,8 +115,7 @@ def set_enable(self, enable_password: str | SecretStr) -> str: if re.search(r">$", current_prompt): enable_output = self.command("enable") if re.search("Password", enable_output): - send_pwd = self.command(pwd) - return send_pwd + return self.command(pwd) elif re.search(r"#$", current_prompt): return "Action: None. Already in enable mode." else: @@ -225,9 +221,7 @@ def connect(self) -> telnetlib.Telnet: OSError: If connection fails """ self.access = telnetlib.Telnet(self.device_name, self.port) - login_prompt = self.access.read_until( - b"(Username: )|(login: )", self.delay - ) + login_prompt = self.access.read_until(b"(Username: )|(login: )", self.delay) if b"login" in login_prompt: self.is_nexus = True self.access.write(self.username.encode("ascii") + b"\n") @@ -271,10 +265,9 @@ def set_enable(self, enable_password: str | SecretStr) -> str: self.access.read_until(b"Password") self.access.write(pwd.encode("ascii") + b"\n") return "Entered enable mode" - elif re.search(b"#$", current_prompt): + if re.search(b"#$", current_prompt): return "Action: None. Already in enable mode." - else: - return "Error: Unable to determine user privilege status." + return "Error: Unable to determine user privilege status." def disable_paging(self, command: str = "term len 0") -> bytes: """Disable paging on the device. diff --git a/netlib/models.py b/netlib/models.py index 9ed0e7e..e6f459f 100644 --- a/netlib/models.py +++ b/netlib/models.py @@ -1,7 +1,5 @@ """Pydantic models for input and output validation.""" -from typing import Any - from pydantic import BaseModel, Field, SecretStr, field_validator @@ -76,9 +74,7 @@ class EnableModeResponse(BaseModel): """Response from enable mode operations.""" message: str = Field(..., description="Response message") - already_enabled: bool = Field( - default=False, description="Whether already in enable mode" - ) + already_enabled: bool = Field(default=False, description="Whether already in enable mode") success: bool = Field(default=True, description="Whether operation succeeded") model_config = {"frozen": False} diff --git a/pyproject.toml b/pyproject.toml index 21570e4..8ba52fc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -82,6 +82,7 @@ ignore = [ [tool.ruff.lint.per-file-ignores] "tests/*" = ["ARG", "PLR2004"] +"tests/test_imports.py" = ["ARG", "PLR2004", "PLC0415"] [tool.mypy] python_version = "3.10" diff --git a/tests/test_imports.py b/tests/test_imports.py index b366958..b3c5a9c 100644 --- a/tests/test_imports.py +++ b/tests/test_imports.py @@ -1,6 +1,6 @@ """Tests for module imports.""" -import pytest +# ruff: noqa: PLC0415 class TestImports: @@ -26,7 +26,7 @@ def test_import_telnet(self) -> None: def test_import_from_package(self) -> None: """Test importing from main package.""" - from netlib import SSH, Telnet, KeyRing + from netlib import SSH, KeyRing, Telnet assert SSH is not None assert Telnet is not None @@ -35,11 +35,11 @@ def test_import_from_package(self) -> None: def test_import_models(self) -> None: """Test importing Pydantic models.""" from netlib.models import ( - SSHConnectionConfig, - TelnetConnectionConfig, CommandResponse, CredentialsData, EnableModeResponse, + SSHConnectionConfig, + TelnetConnectionConfig, ) assert SSHConnectionConfig is not None diff --git a/tests/test_keyring.py b/tests/test_keyring.py index ecbe47d..196466b 100644 --- a/tests/test_keyring.py +++ b/tests/test_keyring.py @@ -1,8 +1,6 @@ """Tests for KeyRing credential management.""" -from unittest.mock import MagicMock, Mock, patch - -import pytest +from unittest.mock import Mock, patch from netlib.user_keyring import KeyRing diff --git a/tests/test_ssh.py b/tests/test_ssh.py index 7f6875c..9d73896 100644 --- a/tests/test_ssh.py +++ b/tests/test_ssh.py @@ -2,8 +2,6 @@ from unittest.mock import MagicMock, Mock, patch -import pytest - from netlib.conn_type import SSH @@ -47,9 +45,7 @@ class TestSSHConnect: @patch("netlib.conn_type.paramiko.SSHClient") @patch("netlib.conn_type.time.sleep") - def test_connect_success( - self, mock_sleep: Mock, mock_ssh_client: Mock - ) -> None: + def test_connect_success(self, mock_sleep: Mock, mock_ssh_client: Mock) -> None: """Test successful SSH connection.""" # Setup mocks mock_client = MagicMock() diff --git a/tests/test_telnet.py b/tests/test_telnet.py index a0750d5..7c1ec1a 100644 --- a/tests/test_telnet.py +++ b/tests/test_telnet.py @@ -2,8 +2,6 @@ from unittest.mock import MagicMock, Mock, patch -import pytest - from netlib.conn_type import Telnet @@ -99,9 +97,7 @@ def test_commands_with_list(self) -> None: """Test executing multiple commands from list.""" telnet = Telnet("switch1", "admin", "secret") - with patch.object( - telnet, "command", return_value=b"output\n" - ) as mock_cmd: + with patch.object(telnet, "command", return_value=b"output\n") as mock_cmd: result = telnet.commands(["show version", "show interfaces"]) assert mock_cmd.call_count == 2 @@ -111,9 +107,7 @@ def test_commands_with_string(self) -> None: """Test executing single command as string.""" telnet = Telnet("switch1", "admin", "secret") - with patch.object( - telnet, "command", return_value=b"output\n" - ) as mock_cmd: + with patch.object(telnet, "command", return_value=b"output\n") as mock_cmd: result = telnet.commands("show version") mock_cmd.assert_called_once_with("show version") From 098da6ce7098d0741e8ef7b3cd6478e4997bff96 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 29 Jan 2026 02:43:01 +0000 Subject: [PATCH 4/5] Fix all linting and testing issues, add Python 3.13 support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Type Checking Fixes (mypy) - Add SecretStr() wrapper when creating Pydantic models - Import SecretStr in user_keyring.py - Convert passwords to SecretStr in SSH/Telnet constructors - Add explicit type annotations for recv() calls to satisfy mypy - Replace deprecated --strict-concatenate with --extra-checks - Fix unused type: ignore comments in __init__.py ## Code Quality Fixes - Fix sendall() to accept bytes instead of strings (encode strings) - Remove unnecessary UTF-8 encoding argument (ruff UP012) - Update pylint configuration to disable false positives: - no-member (Pydantic SecretStr false positives) - deprecated-module (telnetlib is needed despite deprecation) - broad-exception-caught (acceptable for version fallback) - possibly-used-before-assignment (false positive with while loops) - too-many-positional-arguments - redefined-builtin (__name__ is intentional) ## Test Fixes - Update SSH tests to expect bytes in sendall() assertions - Fix test_get_creds_creates_new_when_none_exist mock side_effect (added extra values for nested get_creds() call) - All 54 tests now passing โœ“ ## Python 3.13 Support - Add Python 3.13 to pyproject.toml classifiers - Add Python 3.13 to GitHub Actions test matrix - Note: telnetlib deprecated warning expected (removal in 3.13) ## Verification Results โœ“ ruff format --check: All files properly formatted โœ“ ruff check: All checks passed โœ“ mypy netlib: Success, no issues found โœ“ pylint netlib: 10.00/10 rating โœ“ pytest: 54 tests passed, 82.52% coverage https://claude.ai/code/session_019D1TUrg1xPGse1HsS8pgqQ --- .github/workflows/ci.yml | 2 +- netlib/__init__.py | 4 ++-- netlib/conn_type.py | 14 ++++++++------ netlib/user_keyring.py | 9 +++++---- pyproject.toml | 9 ++++++++- tests/test_keyring.py | 6 ++++-- tests/test_ssh.py | 6 +++--- 7 files changed, 31 insertions(+), 19 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1b63538..84b95a6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -59,7 +59,7 @@ jobs: strategy: fail-fast: false matrix: - python-version: ['3.10', '3.11', '3.12'] + python-version: ['3.10', '3.11', '3.12', '3.13'] steps: - name: Check out code diff --git a/netlib/__init__.py b/netlib/__init__.py index b852a13..4732b7d 100644 --- a/netlib/__init__.py +++ b/netlib/__init__.py @@ -16,11 +16,11 @@ import tomllib else: try: - import tomli as tomllib # type: ignore[import-not-found,no-redef] + import tomli as tomllib # type: ignore[import-not-found] except ImportError: # If tomli not available, use fallback version __version__ = "0.2.0-dev" - tomllib = None # type: ignore[assignment] + tomllib = None # type: ignore[assignment,unused-ignore] if tomllib is not None: pyproject_path = Path(__file__).parent.parent / "pyproject.toml" diff --git a/netlib/conn_type.py b/netlib/conn_type.py index 36e028b..8a65454 100644 --- a/netlib/conn_type.py +++ b/netlib/conn_type.py @@ -39,7 +39,7 @@ def __init__( config = SSHConnectionConfig( device_name=device_name, username=username, - password=password, + password=SecretStr(password), buffer=int(buffer), delay=float(delay), port=int(port), @@ -77,7 +77,8 @@ def connect(self) -> bytes: self.client_conn = self.pre_conn.invoke_shell() time.sleep(self.delay) if self.client_conn: - return self.client_conn.recv(self.buffer) + data: bytes = self.client_conn.recv(self.buffer) + return data return b"" def close(self) -> None: @@ -92,7 +93,8 @@ def clear_buffer(self) -> str | None: Buffer contents if available, None otherwise """ if self.client_conn and self.client_conn.recv_ready(): - return self.client_conn.recv(self.buffer).decode("utf-8", "ignore") + data: bytes = self.client_conn.recv(self.buffer) + return data.decode("utf-8", "ignore") return None def set_enable(self, enable_password: str | SecretStr) -> str: @@ -129,7 +131,7 @@ def disable_paging(self, command: str = "term len 0") -> None: command: Command to disable paging (default: 'term len 0') """ if self.client_conn: - self.client_conn.sendall(f"{command}\n") + self.client_conn.sendall(f"{command}\n".encode()) self.clear_buffer() def command(self, command: str) -> str: @@ -144,7 +146,7 @@ def command(self, command: str) -> str: if not self.client_conn: return "" - self.client_conn.sendall(f"{command}\n") + self.client_conn.sendall(f"{command}\n".encode()) not_done = True output = "" while not_done: @@ -197,7 +199,7 @@ def __init__( config = TelnetConnectionConfig( device_name=device_name, username=username, - password=password, + password=SecretStr(password), delay=float(delay), port=int(port), ) diff --git a/netlib/user_keyring.py b/netlib/user_keyring.py index ee03501..e26b640 100644 --- a/netlib/user_keyring.py +++ b/netlib/user_keyring.py @@ -3,6 +3,7 @@ import getpass import keyring +from pydantic import SecretStr from netlib.models import CredentialsData @@ -37,8 +38,8 @@ def get_creds(self) -> dict[str, str]: # Validate with Pydantic creds = CredentialsData( username=self.username, - password=str(user_pass or ""), - enable=str(enable_pass or ""), + password=SecretStr(str(user_pass or "")), + enable=SecretStr(str(enable_pass or "")), ) return { @@ -73,8 +74,8 @@ def set_creds(self) -> None: # Validate with Pydantic before storing creds = CredentialsData( username=self.username, - password=user_password, - enable=enable_password, + password=SecretStr(user_password), + enable=SecretStr(enable_password), ) # Store in keyring diff --git a/pyproject.toml b/pyproject.toml index 8ba52fc..d235080 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,6 +18,7 @@ classifiers = [ "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", "Topic :: System :: Networking", ] packages = [{include = "netlib"}] @@ -97,7 +98,7 @@ warn_redundant_casts = true warn_unused_ignores = true warn_no_return = true strict_equality = true -strict_concatenate = true +extra_checks = true [[tool.mypy.overrides]] module = [ @@ -112,11 +113,17 @@ disable = [ "too-few-public-methods", "too-many-arguments", "too-many-instance-attributes", + "too-many-positional-arguments", "missing-module-docstring", "missing-class-docstring", "missing-function-docstring", "line-too-long", # handled by ruff "invalid-name", + "redefined-builtin", # __name__ is intentional + "broad-exception-caught", # acceptable for version fallback + "deprecated-module", # telnetlib is deprecated but needed + "no-member", # false positives with Pydantic SecretStr + "possibly-used-before-assignment", # false positive with while loops ] [tool.pytest.ini_options] diff --git a/tests/test_keyring.py b/tests/test_keyring.py index 196466b..a8308ec 100644 --- a/tests/test_keyring.py +++ b/tests/test_keyring.py @@ -42,8 +42,10 @@ def test_get_creds_creates_new_when_none_exist( # First call returns None (no creds), subsequent calls return passwords mock_get_password.side_effect = [ None, # First check - no password exists - "newuserpass", # After setting - "newenablepass", # After setting + "newuserpass", # After setting - get_creds called by set_creds + "newenablepass", # After setting - get_creds called by set_creds + "newuserpass", # Final get_creds call by test + "newenablepass", # Final get_creds call by test ] mock_getpass.side_effect = [ "newuserpass", # Enter user password diff --git a/tests/test_ssh.py b/tests/test_ssh.py index 9d73896..11b2996 100644 --- a/tests/test_ssh.py +++ b/tests/test_ssh.py @@ -96,7 +96,7 @@ def test_command_with_connection(self, mock_sleep: Mock) -> None: result = ssh.command("show version") - mock_channel.sendall.assert_called_once_with("show version\n") + mock_channel.sendall.assert_called_once_with(b"show version\n") assert "Version output" in result @@ -163,7 +163,7 @@ def test_disable_paging(self) -> None: with patch.object(ssh, "clear_buffer"): ssh.disable_paging() - mock_channel.sendall.assert_called_once_with("term len 0\n") + mock_channel.sendall.assert_called_once_with(b"term len 0\n") def test_disable_paging_custom_command(self) -> None: """Test disable paging with custom command.""" @@ -174,7 +174,7 @@ def test_disable_paging_custom_command(self) -> None: with patch.object(ssh, "clear_buffer"): ssh.disable_paging("terminal length 0") - mock_channel.sendall.assert_called_once_with("terminal length 0\n") + mock_channel.sendall.assert_called_once_with(b"terminal length 0\n") class TestSSHClose: From 35c8a473f39ec422f1c09e05a142c7bb20514854 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 29 Jan 2026 02:46:32 +0000 Subject: [PATCH 5/5] Remove Python 3.13 support due to telnetlib removal Python 3.13 removed the telnetlib module from the standard library, which is a core dependency for the Telnet class in this package. Until telnetlib is replaced with an alternative implementation, Python 3.13 cannot be supported. Changes: - Remove Python 3.13 from pyproject.toml classifiers - Remove Python 3.13 from GitHub Actions test matrix - Maximum supported Python version is now 3.12 https://claude.ai/code/session_019D1TUrg1xPGse1HsS8pgqQ --- .github/workflows/ci.yml | 2 +- pyproject.toml | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 84b95a6..1b63538 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -59,7 +59,7 @@ jobs: strategy: fail-fast: false matrix: - python-version: ['3.10', '3.11', '3.12', '3.13'] + python-version: ['3.10', '3.11', '3.12'] steps: - name: Check out code diff --git a/pyproject.toml b/pyproject.toml index d235080..3f39cd2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,7 +18,6 @@ classifiers = [ "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", - "Programming Language :: Python :: 3.13", "Topic :: System :: Networking", ] packages = [{include = "netlib"}]