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..4732b7d 100644 --- a/netlib/__init__.py +++ b/netlib/__init__.py @@ -1,7 +1,39 @@ -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' +import sys +from importlib.metadata import PackageNotFoundError, version +from pathlib import Path + +__name__ = "netlib" + +# 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] + except ImportError: + # If tomli not available, use fallback version + __version__ = "0.2.0-dev" + tomllib = None # type: ignore[assignment,unused-ignore] + + 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 + +__all__ = ["SSH", "KeyRing", "Telnet"] diff --git a/netlib/conn_type.py b/netlib/conn_type.py index 605b2d9..8a65454 100644 --- a/netlib/conn_type.py +++ b/netlib/conn_type.py @@ -1,136 +1,317 @@ -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 telnetlib +import time + +import paramiko +from pydantic import SecretStr + +from netlib.models import ( + 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=SecretStr(password), + buffer=int(buffer), + delay=float(delay), + port=int(port), + ) + + self.device_name = config.device_name + self.username = config.username + self._password = config.password # SecretStr + 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.get_secret_value(), + 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: + data: bytes = self.client_conn.recv(self.buffer) + return data + 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 set_enable(self, enable_password): - if self.re.search('>$', self.command('\n')): - enable = self.command('enable') - if self.re.search('Password', enable): - send_pwd = self.command(enable_password) - return send_pwd - elif self.re.search('#$', self.command('\n')): + 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(): + 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: + """Enter privileged/enable mode. + + Args: + 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): + return self.command(pwd) + 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".encode()) + 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".encode()) 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=SecretStr(password), + delay=float(delay), + port=int(port), + ) + + self.device_name = config.device_name + self.username = config.username + self._password = config.password # SecretStr + 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.get_secret_value().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 set_enable(self, enable_password: str | SecretStr) -> str: + """Enter privileged/enable mode. - def clear_buffer(self): - pass + Args: + enable_password: Enable password (can be str or SecretStr) - 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')): + Returns: + Response message from the operation + """ + 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(pwd.encode("ascii") + b"\n") + return "Entered enable mode" + 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. + + Args: + command: Command to disable paging (default: 'term len 0') - def disable_paging(self, command='term len 0'): - self.access.write(command.encode('ascii') + b'\n') - return self.access.read_until(b"\(#\)|\(>\)", self.delay) + 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): - 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 commands(self, commands_list): - output = str() - if list(commands_list): - for command in commands_list: - output += self.command(command) + Args: + command: Command to execute + + 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..e6f459f --- /dev/null +++ b/netlib/models.py @@ -0,0 +1,80 @@ +"""Pydantic models for input and output validation.""" + +from pydantic import BaseModel, Field, SecretStr, 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: 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.""" + + 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: SecretStr = Field(..., description="User password") + enable: SecretStr = Field(..., description="Enable/privileged password") + + model_config = {"frozen": False} + + @field_validator("password", "enable") + @classmethod + def validate_password_field(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 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..e26b640 100644 --- a/netlib/user_keyring.py +++ b/netlib/user_keyring.py @@ -1,64 +1,104 @@ -class KeyRing(object): +"""User credential management using system keyring.""" - def __init__(self, username): - import keyring - import getpass +import getpass + +import keyring +from pydantic import SecretStr + +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=SecretStr(str(user_pass or "")), + enable=SecretStr(str(enable_pass or "")), + ) + + return { + "username": creds.username, + "password": creds.password.get_secret_value(), + "enable": creds.enable.get_secret_value(), + } + + 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=SecretStr(user_password), + enable=SecretStr(enable_password), + ) + + # Store in keyring + 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() - 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..3f39cd2 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,150 @@ +[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" +tomli = {version = "^2.2.1", python = "<3.11"} + +[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"] +"tests/test_imports.py" = ["ARG", "PLR2004", "PLC0415"] + +[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 +extra_checks = 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", + "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] +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..b3c5a9c --- /dev/null +++ b/tests/test_imports.py @@ -0,0 +1,49 @@ +"""Tests for module imports.""" + +# ruff: noqa: PLC0415 + + +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, KeyRing, Telnet + + 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 ( + CommandResponse, + CredentialsData, + EnableModeResponse, + SSHConnectionConfig, + TelnetConnectionConfig, + ) + + 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..a8308ec --- /dev/null +++ b/tests/test_keyring.py @@ -0,0 +1,178 @@ +"""Tests for KeyRing credential management.""" + +from unittest.mock import Mock, patch + +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 - 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 + "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..6a084f7 --- /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.get_secret_value() == "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.get_secret_value() == "userpass" + assert creds.enable.get_secret_value() == "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..11b2996 --- /dev/null +++ b/tests/test_ssh.py @@ -0,0 +1,196 @@ +"""Tests for SSH connection class.""" + +from unittest.mock import MagicMock, Mock, patch + +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.get_secret_value() == "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(b"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(b"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(b"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..7c1ec1a --- /dev/null +++ b/tests/test_telnet.py @@ -0,0 +1,166 @@ +"""Tests for Telnet connection class.""" + +from unittest.mock import MagicMock, Mock, patch + +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.get_secret_value() == "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