shark/netool-newapp
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Functional read-only app

Browse Source
This commit is contained in:
Cole Deck
2024-05-31 17:35:02 -05:00
parent ed690495d2
commit 1a019a2343
20 changed files with 624 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -7,3 +7,4 @@ listing.txt
__pycache__
build
*.crt
venv

2
config.yml
View File

@@ -1,2 +1,2 @@
tool_directory: ./firmware/www/tool/
app_config_directory: .
app_config_directory: netoolclient/src/netoolclient/resources

3
extract.sh
View File

@@ -16,8 +16,5 @@ mkdir -p $MOUNTPOINT
# Extract the files we need
7z x -o$MOUNTPOINT $IMAGE www/tool/ etc/*.crt -bsp1 -bso0 -bse0 -y
echo "Extracting certificate..."
cp $MOUNTPOINT/etc/*.crt ./apicert.crt
# Perform the diff and show only the differences
#diff --no-dereference -r $MOUNTPOINT

62
netoolclient/.gitignore vendored Normal file
View File

@@ -0,0 +1,62 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# OSX useful to ignore
*.DS_Store
.AppleDouble
.LSOverride
# 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
# C extensions
*.so
# Distribution / packaging
.Python
env/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
*.dist-info/
*.egg-info/
.installed.cfg
*.egg
# IntelliJ Idea family of suites
.idea
*.iml
## File-based project format:
*.ipr
*.iws
## mpeltonen/sbt-idea plugin
.idea_modules/
# Briefcase log files
logs/

5
netoolclient/CHANGELOG Normal file
View File

@@ -0,0 +1,5 @@
# Netool Client Release Notes
## 0.0.1 (31 May 2024)
* Initial release

14
netoolclient/LICENSE Normal file
View File

@@ -0,0 +1,14 @@
Netool Client: A third-party cross-platform Netool companion app.
Copyright (C) 2024 Amelia Deck
This program is free software: you can redistribute it and/or modify
it under the terms of version 3 of the GNU General Public License as
published by the Free Software Foundation.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.

8
netoolclient/README.rst Normal file
View File

@@ -0,0 +1,8 @@
Netool Client
=============
A third-party cross-platform Netool companion app.
.. _`Briefcase`: https://briefcase.readthedocs.io/
.. _`The BeeWare Project`: https://beeware.org/
.. _`becoming a financial member of BeeWare`: https://beeware.org/contributing/membership

184
netoolclient/pyproject.toml Normal file
View File

@@ -0,0 +1,184 @@
# This project was generated with 0.3.18 using template: https://github.com/beeware/briefcase-template@v0.3.18
[tool.briefcase]
project_name = "Netool Client"
bundle = "sh.deck"
version = "0.0.1"
url = "https://git.deck.sh/shark/netool-newapp"
license = "GNU General Public License v3 (GPLv3)"
author = "Amelia Deck"
author_email = "amelia@deck.sh"
[tool.briefcase.app.netoolclient]
formal_name = "Netool Client"
description = "A third-party cross-platform Netool companion app."
long_description = """More details about the app should go here.
"""
sources = [
"src/netoolclient",
]
test_sources = [
"tests",
]
requires = [
"httpx",
]
test_requires = [
"pytest",
]
[tool.briefcase.app.netoolclient.macOS]
universal_build = true
requires = [
"toga-cocoa~=0.4.0",
"std-nslog~=1.0.0",
]
[tool.briefcase.app.netoolclient.linux]
requires = [
"toga-gtk~=0.4.0",
]
[tool.briefcase.app.netoolclient.linux.system.debian]
system_requires = [
# Needed to compile pycairo wheel
"libcairo2-dev",
# Needed to compile PyGObject wheel
"libgirepository1.0-dev",
]
system_runtime_requires = [
# Needed to provide GTK and its GI bindings
"gir1.2-gtk-3.0",
"libgirepository-1.0-1",
# Dependencies that GTK looks for at runtime
"libcanberra-gtk3-module",
# Needed to provide WebKit2 at runtime
# Note: Debian 11 and Ubuntu 20.04 require gir1.2-webkit2-4.0 instead
# "gir1.2-webkit2-4.1",
]
[tool.briefcase.app.netoolclient.linux.system.rhel]
system_requires = [
# Needed to compile pycairo wheel
"cairo-gobject-devel",
# Needed to compile PyGObject wheel
"gobject-introspection-devel",
]
system_runtime_requires = [
# Needed to support Python bindings to GTK
"gobject-introspection",
# Needed to provide GTK
"gtk3",
# Dependencies that GTK looks for at runtime
"libcanberra-gtk3",
# Needed to provide WebKit2 at runtime
# "webkit2gtk3",
]
[tool.briefcase.app.netoolclient.linux.system.suse]
system_requires = [
# Needed to compile pycairo wheel
"cairo-devel",
# Needed to compile PyGObject wheel
"gobject-introspection-devel",
]
system_runtime_requires = [
# Needed to provide GTK
"gtk3",
# Needed to support Python bindings to GTK
"gobject-introspection", "typelib(Gtk) = 3.0",
# Dependencies that GTK looks for at runtime
"libcanberra-gtk3-module",
# Needed to provide WebKit2 at runtime
# "libwebkit2gtk3", "typelib(WebKit2)",
]
[tool.briefcase.app.netoolclient.linux.system.arch]
system_requires = [
# Needed to compile pycairo wheel
"cairo",
# Needed to compile PyGObject wheel
"gobject-introspection",
# Runtime dependencies that need to exist so that the
# Arch package passes final validation.
# Needed to provide GTK
"gtk3",
# Dependencies that GTK looks for at runtime
"libcanberra",
# Needed to provide WebKit2
# "webkit2gtk",
]
system_runtime_requires = [
# Needed to provide GTK
"gtk3",
# Needed to provide PyGObject bindings
"gobject-introspection-runtime",
# Dependencies that GTK looks for at runtime
"libcanberra",
# Needed to provide WebKit2 at runtime
# "webkit2gtk",
]
[tool.briefcase.app.netoolclient.linux.appimage]
manylinux = "manylinux_2_28"
system_requires = [
# Needed to compile pycairo wheel
"cairo-gobject-devel",
# Needed to compile PyGObject wheel
"gobject-introspection-devel",
# Needed to provide GTK
"gtk3-devel",
# Dependencies that GTK looks for at runtime, that need to be
# in the build environment to be picked up by linuxdeploy
"libcanberra-gtk3",
"PackageKit-gtk3-module",
"gvfs-client",
]
linuxdeploy_plugins = [
"DEPLOY_GTK_VERSION=3 gtk",
]
[tool.briefcase.app.netoolclient.linux.flatpak]
flatpak_runtime = "org.gnome.Platform"
flatpak_runtime_version = "45"
flatpak_sdk = "org.gnome.Sdk"
[tool.briefcase.app.netoolclient.windows]
requires = [
"toga-winforms~=0.4.0",
]
# Mobile deployments
[tool.briefcase.app.netoolclient.iOS]
requires = [
"toga-iOS~=0.4.0",
"std-nslog~=1.0.0",
]
[tool.briefcase.app.netoolclient.android]
requires = [
"toga-android~=0.4.0",
]
base_theme = "Theme.MaterialComponents.Light.DarkActionBar"
build_gradle_dependencies = [
"androidx.appcompat:appcompat:1.6.1",
"com.google.android.material:material:1.11.0",
# Needed for DetailedList
"androidx.swiperefreshlayout:swiperefreshlayout:1.1.0",
]
# Web deployments
[tool.briefcase.app.netoolclient.web]
requires = [
"toga-web~=0.4.0",
]
style_framework = "Shoelace v2.3"

0
netoolclient/src/netoolclient/__init__.py Normal file
View File

4
netoolclient/src/netoolclient/__main__.py Normal file
View File

@@ -0,0 +1,4 @@
from netoolclient.app import main
if __name__ == "__main__":
main().main_loop()

272
netoolclient/src/netoolclient/app.py Normal file
View File

@@ -0,0 +1,272 @@
"""
A third-party cross-platform Netool companion app.
"""
import toga
from toga.style import Pack
from toga.style.pack import COLUMN, LEFT, RIGHT, TOP, BOTTOM, CENTER, ROW, Pack
from netoolclient.call_api import call_api_preloaded
import asyncio
import threading
import json
import socket
def run_asyncio_event_loop():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_forever()
class NetoolClient(toga.App):
ip = "192.168.49.1"
conntype = "eth" # or ble
connected = False
page = "Status"
lastpage = "None"
page_mode = "DetailedList" # Text, Switches, etc
refresh = True
def startup(self):
"""Construct and show the Toga application.
Usually, you would add your application to a main content box.
We then create a main window (with a name matching the app), and
show the main window.
"""
self.main_box = toga.Box(style=Pack(direction=COLUMN))
self.status_header = toga.Label(
"Connecting to Netool.IO device via WiFi...",
style=Pack(padding=(5, 5)),
)
self.page_box = toga.Box(style=Pack(direction=ROW, padding=(5,5)))
# self.page_header = toga.Label(
# "",
# style=Pack(padding=(5, 5)),
# )
self.page_selector = toga.Selection(items=["Status", "Discovery Packet (LLDP, CDP...) Details", "STP Details", "ARP Scan", "Traceroute Log", "NTP Status", "PCAP Status", "History", "WiFi Settings", "Ethernet Settings", "Discovery Timers", "Scan Toggle Switches", "About", ], style=Pack(flex=1), on_change=self.set_refresh)
self.page_box.add(self.page_selector)
self.page_loading = toga.ProgressBar(max=None, style=Pack(alignment=RIGHT, flex=1))
self.info_box = toga.Box(style=Pack(direction=COLUMN, flex=1))
# self.info_text = toga.Label(
# "",
# style=Pack(padding=(5, 5), flex=1),
# )
self.info_text = toga.MultilineTextInput(readonly=True, value="", style=Pack(padding=(5, 5), flex=1))
self.info_list = toga.DetailedList(data=[], style=Pack(padding=(5, 5), flex=1))
self.info_box.add(self.info_list)
# container = toga.ScrollContainer(content=self.info_box, style=Pack(flex=1), horizontal=False)
self.main_box.add(self.status_header)
# self.page_box.add(self.page_header)
self.main_box.add(self.page_loading)
self.main_box.add(self.page_box)
self.main_box.add(self.info_box)
if toga.platform.current_platform == "android":
self.page_refresh = toga.Button(text="Refresh", style=Pack(flex=5, alignment=CENTER), on_press=self.set_refresh)
self.auto_refresh = toga.Switch(text=None, style=Pack(padding_top=11, flex=1, alignment=CENTER))
else:
self.page_refresh = toga.Button(text="Refresh", style=Pack(flex=7, alignment=CENTER), on_press=self.set_refresh)
self.auto_refresh = toga.Switch(text="Auto", style=Pack(padding_top=4, padding_left=8, flex=1, alignment=CENTER))
self.refresh_box = toga.Box(style=Pack(direction=ROW, padding=(5,5)))
self.refresh_box.add(self.page_refresh)
self.refresh_box.add(self.auto_refresh)
self.main_box.add(self.refresh_box)
self.main_window = toga.MainWindow(title=self.formal_name)
self.main_window.content = self.main_box
self.main_window.show()
#toga.App.add_background_task(self, handler=self.check_online)
toga.App.add_background_task(self, handler=self.update_page)
def set_refresh(self, a):
self.refresh = True
async def check_online(self):
# Your periodic task logic here
if True or self.conntype == "eth":
result = await self.check_online_wifi()
if result:
text = "Connected to Netool.IO device."
self.connected = True
else:
text = "Connecting to Netool.IO device via WiFi..."
self.connected = False
#print(text)
# Schedule the GUI update on the main thread
if self.status_header.text != text:
print(text)
self.status_header.text = text
# Wait for 5 seconds before running again
return result
async def update_page(self, a):
counter = 0
while True:
self.page = self.page_selector.value
if self.connected:
if (self.refresh) and await self.check_online():
self.refresh = False
print("Switching to page " + self.page)
self.lastpage = self.page
#self.page_header.text = self.page
self.page_loading.start()
data = await self.get_info()
if self.page_mode == "Text":
self.info_text.value = str(data)
if self.page_mode == "DetailedList":
self.info_list.data = data
self.page_loading.stop()
else:
await self.check_online()
await asyncio.sleep(0.5)
if self.auto_refresh.value == True:
counter += 1
if counter >= 4:
counter = 0
self.refresh=True
else:
counter = 0
# render new page
else:
result = await self.check_online()
if not result:
await asyncio.sleep(0.25)
def set_info_mode(self, mode):
if mode != self.page_mode:
match self.page_mode:
case "DetailedList":
self.info_box.remove(self.info_list)
case "Text":
self.info_box.remove(self.info_text)
self.page_mode = mode
match mode:
case "DetailedList":
self.info_box.add(self.info_list)
case "Text":
self.info_box.add(self.info_text)
async def get_info(self):
match self.page:
case "Status":
call = "operations"
case "About":
call = "about"
case "History":
call = "history"
case "ARP Scan":
call = "arpscan"
case "Discovery Packet (LLDP, CDP...) Details":
call = "dp_detail"
case "STP Details":
call = "stp_detail"
case "NTP Status":
call = "ntp_status"
case "PCAP Status":
call = "sniff"
case "Discovery Timers":
call = "scan_time_status"
case "WiFi Settings":
call = "wifi_status"
case "Ethernet Settings":
call = "eth_status"
case "Scan Toggle Switches":
call = "toggle_settings_status"
case "Traceroute Log":
call = "traceroute_log"
case _:
call = None
if call is not None:
res = await self.call_api_safe(call)
while res is None:
res = await self.call_api_safe(call)
#return res
match self.page:
case "Status":
self.set_info_mode("DetailedList")
return [{"title": key, "subtitle": str(val).replace("\n", " ")} for key, val in res.items()]
case "About":
self.set_info_mode("DetailedList")
return [{"title": key, "subtitle": str(val).replace("\n", " ")} for key, val in res.items()]
case "History":
self.set_info_mode("DetailedList")
out = []
for index, entry in enumerate(res):
for key, val in entry.items():
out.append({"title": str(index)+" "+key, "subtitle": str(val).replace("\n", " ")})
print(out)
return out
case "ARP Scan":
self.set_info_mode("DetailedList")
return [{"title": dev["IP"], "subtitle": dev["MAC"]} for dev in res]
case "Discovery Packet (LLDP, CDP...) Details":
self.set_info_mode("Text")
return res.replace("<br>", "\n")
case "STP Details":
self.set_info_mode("Text")
return res.replace("<br>", "\n")
case "NTP Status":
self.set_info_mode("DetailedList")
return [{"title": key, "subtitle": str(val).replace("_", " ")} for key, val in res.items()]
case "PCAP Status":
self.set_info_mode("DetailedList")
return [{"title": key, "subtitle": str(val).replace("\n", " ")} for key, val in res.items()]
case "Discovery Timers":
self.set_info_mode("DetailedList")
return [{"title": key, "subtitle": str(val).replace("\n", " ") + " seconds"} for key, val in res.items()]
case "WiFi Settings":
self.set_info_mode("DetailedList")
return [{"title": key, "subtitle": str(val).replace("\n", " ")} for key, val in res.items()]
case "Ethernet Settings":
self.set_info_mode("DetailedList")
return [{"title": key, "subtitle": str(val).replace("\n", " ")} for key, val in res.items()]
case "Scan Toggle Switches":
self.set_info_mode("DetailedList")
return [{"title": key, "subtitle": str(val).replace("\n", " ")} for key, val in res.items()]
case "Traceroute Log":
self.set_info_mode("Text")
return res.replace("<br><br>", "\n")
async def call_api_safe(self, callname, method="auto", params={}, baseaddr="https://192.168.49.1/tool/"):
try:
response = await call_api_preloaded(callname, method, params, baseaddr, conntype=self.conntype, filepath=str(self.paths.app) + "/resources")
#print("Response Status Code:", response.status_code)
try:
#print("Response JSON:", json.dumps(response.json(), indent=2))
return response.json()
except ValueError:
#print("Response Text:", response.text)
return response.text
except Exception as e:
print("Error:", str(e))
return None
async def check_online_wifi(self):
# host = await async_ping(self.ip, count=1, timeout=0.25)
# return host.is_alive
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
result = sock.connect_ex((self.ip,443))
sock.close()
return result == 0
def main():
return NetoolClient()

32
call_api.py → netoolclient/src/netoolclient/call_api.py
View File

@@ -1,7 +1,11 @@
import requests
from read_api_details import parse_csv_to_dict
import httpx
try:
from netoolclient.read_api_details import parse_csv_to_dict
except:
from read_api_details import parse_csv_to_dict
from sys import platform
import subprocess
import asyncio
def ping(host):
@@ -21,7 +25,7 @@ def ping(host):
def call_api(api_url, method, params):
async def call_api(api_url, method, params, filepath):
"""
Function to make an API call.
@@ -36,16 +40,20 @@ def call_api(api_url, method, params):
print("Calling API",api_url,"with method",method,"and parameters",params)
if method.upper() == 'POST':
response = requests.post(api_url, data=params, verify='./apicert.crt', timeout=8)
# response = requests.post(api_url, data=params, verify=filepath + '/apicert.crt', timeout=8)
async with httpx.AsyncClient(verify=filepath + '/apicert.crt') as client:
response = await client.post(api_url, data=params, timeout=8)
elif method.upper() == 'GET':
response = requests.get(api_url, params=params, verify='./apicert.crt', timeout=8)
# response = requests.get(api_url, params=params, verify=filepath + './apicert.crt', timeout=8)
async with httpx.AsyncClient(verify=filepath + '/apicert.crt') as client:
response = await client.get(api_url, params=params, timeout=8)
else:
raise ValueError("Method must be 'POST' or 'GET'")
return response
def call_api_preloaded(callname, method="auto", params={}, baseaddr="https://192.168.49.1/tool/", conntype="eth"):
details = parse_csv_to_dict("apidetails.csv")
async def call_api_preloaded(callname, method="auto", params={}, baseaddr="https://192.168.49.1/tool/", conntype="eth", filepath="resources"):
details = parse_csv_to_dict(filepath + "/apidetails.csv")
if not callname.find(".php") > 0:
callname += ".php"
full_url = baseaddr + callname
@@ -90,8 +98,8 @@ def call_api_preloaded(callname, method="auto", params={}, baseaddr="https://192
for param in auto_include:
params[param[0]] = param[1]
if not ping("192.168.49.1"):
print("Connecting to netool...")
while not ping("192.168.49.1"):
pass
return call_api(full_url, method, params)
# if not ping("192.168.49.1"):
# print("Connecting to netool...")
# while not ping("192.168.49.1"):
# pass
return await call_api(full_url, method, params, filepath)

5
interactive_api.py → netoolclient/src/netoolclient/interactive_api.py
View File

@@ -1,9 +1,10 @@
from call_api import *
from read_api_details import parse_csv_to_dict
import json
import asyncio
def main():
details = parse_csv_to_dict("apidetails.csv")
details = parse_csv_to_dict("resources/apidetails.csv")
print(details.keys())
# Prompt the user for the API call
api_call = input("Enter the API call (e.g., about.php): ").strip()
@@ -24,7 +25,7 @@ def main():
# Call the API
try:
response = call_api_preloaded(api_call, method, params)
response = asyncio.run(call_api_preloaded(api_call, method, params))
print("Response Status Code:", response.status_code)
try:
print("Response JSON:", json.dumps(response.json(), indent=2))

0
read_api_details.py → netoolclient/src/netoolclient/read_api_details.py
View File

2
netoolclient/src/netoolclient/resources/README Normal file
View File

@@ -0,0 +1,2 @@
Put any application resources (e.g., icons and resources) here;
they can be referenced in code as "resources/filename".

0
netoolclient/tests/__init__.py Normal file
View File

35
netoolclient/tests/netoolclient.py Normal file
View File

@@ -0,0 +1,35 @@
import os
import sys
import tempfile
from pathlib import Path
import pytest
def run_tests():
project_path = Path(__file__).parent.parent
os.chdir(project_path)
# Determine any args to pass to pytest. If there aren't any,
# default to running the whole test suite.
args = sys.argv[1:]
if len(args) == 0:
args = ["tests"]
returncode = pytest.main(
[
# Turn up verbosity
"-vv",
# Disable color
"--color=no",
# Overwrite the cache directory to somewhere writable
"-o",
f"cache_dir={tempfile.gettempdir()}/.pytest_cache",
] + args
)
print(f">>>>>>>>>> EXIT {returncode} <<<<<<<<<<")
if __name__ == "__main__":
run_tests()

3
netoolclient/tests/test_app.py Normal file
View File

@@ -0,0 +1,3 @@
def test_first():
"""An initial test for the app."""
assert 1 + 1 == 2

5
requirements.txt Normal file
View File

@@ -0,0 +1,5 @@
# requirements for GUI app
# not needed for core API
briefcase
httpx
icmplib

9
setup-api.sh
View File

@@ -2,7 +2,7 @@
set -euo pipefail
if (! [ -e apidetails.csv ]) || (! [ -e apicert.crt ]); then
if (! [ -e netoolclient/src/netoolclient/resources/apidetails.csv ]) || (! [ -e netoolclient/src/netoolclient/resources/apicert.crt ]); then
echo "Downloading firmware for device..."
./download-fw.sh
@@ -12,8 +12,9 @@ if (! [ -e apidetails.csv ]) || (! [ -e apicert.crt ]); then
DIR=./firmware/www/tool/
echo "tool_directory: $DIR
app_config_directory: ." > config.yml
app_config_directory: netoolclient/src/netoolclient/resources" > config.yml
echo "Extracting certificate..."
cp ./firmware/etc/*.crt netoolclient/src/netoolclient/resources/apicert.crt
echo "Extracting API keys..."
python get_codes.py > /dev/null
echo "Cleaning up..."
@@ -21,4 +22,4 @@ app_config_directory: ." > config.yml
echo "API client is setup."
else
echo "Already setup."
fi
fi