DAProgs/portspoof_py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

2603.2 - change password via web

Browse Source
This commit is contained in:
DAProgs
2026-03-11 19:13:35 -04:00
parent c2624d165a
commit ee77eb23c8
4 changed files with 103 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
portspoof_py/__init__.py
View File

@@ -1,2 +1,2 @@
"""portspoof_py — asyncio Python rewrite of the portspoof TCP honeypot.""" """portspoof_py — asyncio Python rewrite of the portspoof TCP honeypot."""
__version__ = '2603.1' __version__ = '2603.2'

103
portspoof_py/admin.py
View File

@@ -71,6 +71,63 @@ _UNAUTHORIZED = (
b'Unauthorized' b'Unauthorized'
) )
# ── passwd page ───────────────────────────────────────────────────────────────
_PASSWD_HTML = """\
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>portspoof — change password</title>
<style>{css}</style>
</head>
<body>
<h1>portspoof admin</h1>
<p class="sub"><a href="/">&larr; dashboard</a></p>
{banner_html}
<div class="card">
<h2>Change password</h2>
<form method="post" action="/api/passwd">
<div class="field">
<label>Current password</label>
<input type="password" name="current_password" autofocus>
</div>
<div class="field">
<label>New password</label>
<input type="password" name="new_password">
</div>
<div class="field">
<label>Confirm new password</label>
<input type="password" name="confirm_password">
</div>
<div class="actions">
<button type="submit">Update password</button>
</div>
</form>
</div>
</body>
</html>
"""
_PASSWD_ERRORS = {
'wrong_current': 'Current password is incorrect.',
'mismatch': 'New passwords do not match.',
'empty': 'New password must not be empty.',
'save': 'Password updated in memory but could not be written to disk.',
}
def _render_passwd(msg: str = '', msg_ok: bool = True) -> str:
banner_html = ''
if msg:
cls = 'ok' if msg_ok else 'err'
banner_html = f'<div class="banner {cls}">{html.escape(msg)}</div>'
return _PASSWD_HTML.format(css=_CONFIG_CSS, banner_html=banner_html)
# ── config page ─────────────────────────────────────────────────────────────── # ── config page ───────────────────────────────────────────────────────────────
_CONFIG_CSS = """ _CONFIG_CSS = """
@@ -335,7 +392,8 @@ _HTML = """\
<button type="submit" class="link-btn">reset stats</button> <button type="submit" class="link-btn">reset stats</button>
</form> </form>
&middot; <a href="/config">email alerts</a> &middot; &middot; <a href="/config">email alerts</a> &middot;
<a href="/api/stats" target="_blank">JSON stats</a> <a href="/api/stats" target="_blank">JSON stats</a> &middot;
<a href="/passwd">change password</a>
</p> </p>
<!-- stat cards --> <!-- stat cards -->
@@ -532,7 +590,8 @@ async def _handle(
writer: asyncio.StreamWriter, writer: asyncio.StreamWriter,
stats: Stats, stats: Stats,
cfg: Config, cfg: Config,
creds: Tuple[str, str], creds_mut: list,
passwd_file: str,
notifier: Optional['Notifier'] = None, notifier: Optional['Notifier'] = None,
) -> None: ) -> None:
try: try:
@@ -543,7 +602,7 @@ async def _handle(
if len(parts) < 2: if len(parts) < 2:
return return
if not _check_auth(text, creds): if not _check_auth(text, creds_mut):
writer.write(_UNAUTHORIZED) writer.write(_UNAUTHORIZED)
await writer.drain() await writer.drain()
return return
@@ -645,6 +704,40 @@ async def _handle(
body = json.dumps(notifier.get_config_safe(), indent=2) body = json.dumps(notifier.get_config_safe(), indent=2)
ct = 'application/json' ct = 'application/json'
# ── password change ───────────────────────────────────────────────────
elif path == '/passwd':
msg = ''
msg_ok = True
if 'ok' in qs:
msg = 'Password updated successfully.'
elif 'err' in qs:
msg = _PASSWD_ERRORS.get(qs['err'][0], 'Unknown error.')
msg_ok = False
body = _render_passwd(msg, msg_ok)
ct = 'text/html; charset=utf-8'
elif path == '/api/passwd' and method == 'POST':
form = _parse_post_body(text)
current = form.get('current_password', '')
new_pw = form.get('new_password', '')
confirm = form.get('confirm_password', '')
_, expected_passwd = creds_mut
if not hmac.compare_digest(current, expected_passwd):
writer.write(_redirect('/passwd?err=wrong_current'))
elif not new_pw:
writer.write(_redirect('/passwd?err=empty'))
elif new_pw != confirm:
writer.write(_redirect('/passwd?err=mismatch'))
else:
creds_mut[1] = new_pw
try:
Path(passwd_file).write_text(f'{creds_mut[0]}:{new_pw}\n')
writer.write(_redirect('/passwd?ok=1'))
except Exception:
writer.write(_redirect('/passwd?err=save'))
await writer.drain()
return
# ── standard routes ─────────────────────────────────────────────────── # ── standard routes ───────────────────────────────────────────────────
elif path == '/api/stats/reset' and method == 'POST': elif path == '/api/stats/reset' and method == 'POST':
stats.reset() stats.reset()
@@ -721,11 +814,13 @@ async def run_admin(
stop_event: asyncio.Event, stop_event: asyncio.Event,
notifier: Optional['Notifier'] = None, notifier: Optional['Notifier'] = None,
ssl_context: Optional[ssl.SSLContext] = None, ssl_context: Optional[ssl.SSLContext] = None,
passwd_file: str = 'admin.passwd',
) -> None: ) -> None:
"""Start the admin HTTP(S) server and run until stop_event is set.""" """Start the admin HTTP(S) server and run until stop_event is set."""
creds_mut = list(creds) # mutable so password changes propagate across requests
def handler(r, w): def handler(r, w):
asyncio.create_task(_handle(r, w, stats, cfg, creds, notifier)) asyncio.create_task(_handle(r, w, stats, cfg, creds_mut, passwd_file, notifier))
server = await asyncio.start_server( server = await asyncio.start_server(
handler, handler,

3
portspoof_py/cli.py
View File

@@ -176,7 +176,8 @@ def main(argv=None) -> int:
if args.admin_port: if args.admin_port:
tasks.append(asyncio.create_task( tasks.append(asyncio.create_task(
run_admin(stats, cfg, args.admin_host, args.admin_port, creds, stop_event, run_admin(stats, cfg, args.admin_host, args.admin_port, creds, stop_event,
notifier=notifier, ssl_context=ssl_context) notifier=notifier, ssl_context=ssl_context,
passwd_file=args.admin_passwd)
)) ))
await asyncio.gather(*tasks) await asyncio.gather(*tasks)
finally: finally:

2
pyproject.toml
View File

@@ -4,7 +4,7 @@ build-backend = "flit_core.buildapi"
[project] [project]
name = "portspoof-py" name = "portspoof-py"
version = "2603.1" version = "2603.2"
description = "Python asyncio rewrite of the portspoof TCP honeypot" description = "Python asyncio rewrite of the portspoof TCP honeypot"
readme = "README.md" readme = "README.md"
requires-python = ">=3.11" requires-python = ">=3.11"