first commit

.devcontainer/Dockerfile

@@ -0,0 +1,9 @@
+FROM python:3.10.2-slim
+
+RUN apt update \
+    && apt install -y --no-install-recommends apt-utils git gcc build-essential \
+    && pip install --no-cache-dir autopep8 flake8 pytest \
+    && apt-get autoremove -y \
+    && apt-get clean -y \
+    && rm -rf /var/lib/apt/lists/*
+

.devcontainer/devcontainer.json

@@ -0,0 +1,35 @@
+{
+    "name": "Python 3",
+    "build": {
+		"dockerfile": "Dockerfile",
+        "context": ".."
+    },
+    "mounts": [
+        "source=/data,target=/data,type=bind,consistency=cached",
+    ],
+    "settings": {
+        "python.pythonPath": "/usr/local/bin/python",
+        "python.linting.pylintEnabled": false,
+        "python.linting.flake8Enabled": true,
+        "python.linting.flake8Args": [
+            "--ignore=E402",
+            "--max-line-length",
+            "4096"
+        ],
+        "python.formatting.provider": "autopep8",
+        "python.formatting.autopep8Args": [
+            "--ignore",
+            "E402",
+            "--max-line-length",
+            "4096"
+        ],
+        "[python]": {
+            "editor.formatOnSave": true
+        }
+    },
+    "extensions": [
+        "ms-python.python",
+        "ms-python.vscode-pylance"
+    ],
+    "postCreateCommand": "pip install -r requirements.txt",
+}

.gitignore

@@ -0,0 +1,213 @@
+
+# Created by https://www.toptal.com/developers/gitignore/api/osx,python,windows
+# Edit at https://www.toptal.com/developers/gitignore?templates=osx,python,windows
+
+### OSX ###
+# General
+.DS_Store
+.AppleDouble
+.LSOverride
+
+# Icon must end with two \r
+Icon
+
+
+# Thumbnails
+._*
+
+# Files that might appear in the root of a volume
+.DocumentRevisions-V100
+.fseventsd
+.Spotlight-V100
+.TemporaryItems
+.Trashes
+.VolumeIcon.icns
+.com.apple.timemachine.donotpresent
+
+# Directories potentially created on remote AFP share
+.AppleDB
+.AppleDesktop
+Network Trash Folder
+Temporary Items
+.apdisk
+
+### Python ###
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.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/
+cover/
+
+# 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
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow
+__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/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+
+### Windows ###
+# Windows thumbnail cache files
+Thumbs.db
+Thumbs.db:encryptable
+ehthumbs.db
+ehthumbs_vista.db
+
+# Dump file
+*.stackdump
+
+# Folder config file
+[Dd]esktop.ini
+
+# Recycle Bin used on file shares
+$RECYCLE.BIN/
+
+# Windows Installer files
+*.cab
+*.msi
+*.msix
+*.msm
+*.msp
+
+# Windows shortcuts
+*.lnk
+

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2022 Yoshihiro OKUMURA (http://github.com/orrisroot/)
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1 @@
+# Switch configuration backup script

requirements.txt

@@ -0,0 +1,2 @@
+paramiko
+paramiko-expect

swcfg-backup.py

@@ -0,0 +1,203 @@
+#!/bin/env python3
+
+import hashlib
+import json
+import os
+import shutil
+import stat
+import telnetlib
+from datetime import datetime
+from typing import List
+from zoneinfo import ZoneInfo
+
+import paramiko
+from paramiko_expect import SSHClientInteraction
+
+
+class JsonConfigLoader:
+
+    def __init__(self, fpath: str):
+        with open(fpath, 'r', encoding='utf-8') as fp:
+            self.data = json.load(fp)
+            fp.close()
+
+    @property
+    def tftp_host(self) -> str:
+        return self.data['tftp']['server']
+
+    @property
+    def tftp_path(self) -> str:
+        return self.data['tftp']['path']
+
+    @property
+    def tftp_rootdir(self) -> str:
+        return self.data['tftp']['rootdir']
+
+    @property
+    def backup_destdir(self) -> str:
+        return self.data['backup']['destdir']
+
+    @property
+    def hosts(self) -> str:
+        return self.data['hosts']
+
+
+class ConnectionBase:
+    TIMEOUT = 30
+    PROMPT_USERNAME = ['User(name)?:\\s*']
+    PROMPT_PASSWORD = ['Pass(word)?:\\s*']
+    PROMPT_COMMAND = ['.*\\]\\s*', '.*>\\s*', '.*#\\s*']
+
+    def __init__(self, hostname: str):
+        self.hostname = hostname
+
+    def login(self, username: str, password: str):
+        self.username = username
+        self.password = password
+
+    def send(self, line: str):
+        pass
+
+    def expect(self, patterns: List):
+        pass
+
+    def wait_command_prompt(self):
+        self.expect(self.PROMPT_COMMAND)
+
+    def close(self):
+        self.username = None
+        self.password = None
+
+
+class ConnectionTelnet(ConnectionBase):
+
+    def login(self, username: str, password: str):
+        self.conn = telnetlib.Telnet(self.hostname, timeout=self.TIMEOUT)
+        self.expect(self.PROMPT_USERNAME)
+        self.send(username)
+        self.expect(self.PROMPT_PASSWORD)
+        self.send(password)
+        self.wait_command_prompt()
+        super().login(username, password)
+
+    def send(self, line: str):
+        # print('send:' + line)
+        self.conn.write(line.encode('utf-8') + b'\n')
+
+    def expect(self, patterns: List):
+        # print('expect:' + ','.join(patterns))
+        self.conn.expect(list(map(lambda x: x.encode('utf-8'), patterns)), self.TIMEOUT)
+
+    def close(self):
+        self.conn.read_all()
+        self.conn.close()
+        self.conn = None
+        super().close()
+
+
+class ConnectionSsh(ConnectionBase):
+
+    def login(self, username: str, password: str):
+        self.client = paramiko.SSHClient()
+        self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        self.client.connect(hostname=self.hostname, username=username, password=password, timeout=self.TIMEOUT, look_for_keys=False)
+        self.conn = SSHClientInteraction(self.client, timeout=self.TIMEOUT, display=False)
+        self.send('')
+        self.wait_command_prompt()
+        super().login(username, password)
+
+    def send(self, line: str):
+        # print('send:' + line)
+        self.conn.send(line)
+
+    def expect(self, patterns: List):
+        # print('expect:' + ','.join(patterns))
+        self.conn.expect(patterns, self.TIMEOUT)
+
+    def close(self):
+        self.client.close()
+        self.conn = None
+        self.client = None
+        super().close()
+
+
+class SwitchConfigFetcher:
+
+    def __init__(self, host: dict[str, str]):
+        self.hostname = host['hostname']
+        self.protocol = host['protocol']
+        self.system = host['system']
+        self.username = host['username']
+        self.password = host['password']
+        self.enable = host['enable'] if 'enable' in host else None
+
+    def fetch(self, tftp_server: str, tftp_fpath: str):
+        conn = ConnectionTelnet(self.hostname) if self.protocol == 'telnet' else ConnectionSsh(self.hostname)
+        conn.login(self.username, self.password)
+        if self.system in ['s5100', 'a5120']:
+            config = 'config.cfg' if self.system == 's5100' else 'startup.cfg'
+            conn.send('tftp ' + tftp_server + ' put ' + config + ' ' + tftp_fpath)
+            conn.wait_command_prompt()
+            conn.send('quit')
+        else:
+            if self.enable is not None:
+                conn.send('enable')
+                if self.enable != "":
+                    conn.expect(ConnectionBase.PROMPT_PASSWORD)
+                    conn.send(self.enable)
+                conn.wait_command_prompt()
+            conn.send('copy running-config tftp://' + tftp_server + '/' + tftp_fpath)
+            if self.system in ['n4000', 'n3000']:
+                conn.expect(['.*\\(y/n\\)\\s*'])
+                conn.send('y')
+            if self.enable is not None:
+              conn.wait_command_prompt()
+              conn.send('exit')
+            conn.wait_command_prompt()
+            conn.send('exit')
+        conn.close()
+
+
+class SwitchConfigRotater:
+
+    def __init__(self, fpath: str):
+        self.fpath = fpath
+
+    def rotate(self, backupdir: str):
+        if not os.path.isdir(backupdir):
+            os.mkdir(backupdir)
+        latest_fpath = os.path.join(backupdir, 'latest.cfg')
+        if os.path.isfile(latest_fpath):
+            hash_latest = self._sha256(latest_fpath)
+            hash_current = self._sha256(self.fpath)
+            if hash_latest != hash_current:
+                mtime = datetime.fromtimestamp(os.stat(latest_fpath).st_mtime, tz=ZoneInfo('Asia/Tokyo'))
+                now = mtime.strftime('%Y%m%d%H%M%S')
+                rotate_fpath = os.path.join(backupdir, now + '.cfg')
+                shutil.move(latest_fpath, rotate_fpath)
+                shutil.copy(self.fpath, latest_fpath)
+                os.chmod(latest_fpath, stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP)
+        else:
+            shutil.copy(self.fpath, latest_fpath)
+            os.chmod(latest_fpath, stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP)
+
+    def _sha256(self, fpath):
+        h = hashlib.sha256()
+        with open(fpath, 'rb') as f:
+            h.update(f.read())
+        return h.hexdigest()
+
+
+def main(fpath: str):
+    config = JsonConfigLoader(fpath)
+    for host in config.hosts:
+        fname = host['hostname'] + '.cfg'
+        scf = SwitchConfigFetcher(host)
+        scf.fetch(config.tftp_host, os.path.join(config.tftp_path, fname))
+        scr = SwitchConfigRotater(os.path.join(config.tftp_rootdir, config.tftp_path, fname))
+        scr.rotate(os.path.join(config.backup_destdir, host['hostname']))
+
+
+if __name__ == '__main__':
+    config_fpath = './swcfg-backup.json'
+    main(config_fpath)

swcfg-backup.sample.json

@@ -0,0 +1,17 @@
+{
+  "tftp": {
+    "server": "192.168.1.100",
+    "path": "config",
+    "rootdir": "/var/lib/tftpboot"
+  },
+  "backup": {
+    "destdir": "/backup/switch"
+  },
+  "hosts": [
+    {"hostname": "s5100", "protocol": "telnet", "system": "s5100", "username": "admin", "password": "secret"},
+    {"hostname": "a5120", "protocol": "telnet", "system": "a5120", "username": "admin", "password": "secret"},
+    {"hostname": "n4032f", "protocol": "telnet", "system": "n4000", "username": "admin", "password": "secret", "enable": "secret"},
+    {"hostname": "n3224t", "protocol": "ssh", "system": "n3000", "username": "admin", "password": "secret", "enable": ""},
+    {"hostname": "s4128f", "protocol": "ssh", "system": "s4000", "username": "admin", "password": "secret"}
+  ]
+}