# Only Stop Learning
## When No Air Remains.
I have mentioned this project that I have been working on, and some not really in much detail. In some cases there is such thing as too much detail, but what I give up here is not too big of a deal. I am running what is essentially a 10 year old gaming rig with an 500 GB SSD, a 1TB spinning rust drive and 16Gb of RAM as a Windows Server 2025 Domain Controller handling Identity and serving DNS to my Unifi DM7 -Pro. That DM7-Pro is running basic intrusion protection, 3 isolated vlans, and a host isolated guest network. There is another router in the mix, but it is configured in pass through mode to prevent double-NAT issues. Moving right on across the room is a 23 year old Dell workstation that I bought years ago when I decided I was going to go to college. This machine is a trooper. I used it for years, sold it, had it given back to me, and moved it around a few times. When I decided that perhaps I should do something with modern day infrastructure or rather applications, because most of my hardware is anything BUT modern, I pushed power and she powered up, so I figured I better give her a job to do. Aside from the dead CMOS battery, and extremely loud variable speed fan, it has been a solid little machine.
Today she runs a pretty impressive stack really, in terms of a home network, this might be overkill, but she just does her job. Right now her job is to host Prometheus, Loki, Grafana, Tempo, and FleetDM. Why in the world would someone fight obsolete hardware to install such a setup? Well, it goes like this. A college degree and certifications will never open the doors for you that raw practical experience will. The experience alone is not enough, nor is either one of the other two mentioned factors.
The damning truth is that unless you take the initiative to fight your way through and build it, you may never get the experience that you need. Back to the task. Because this machine is 23 years old, each of these packages have to run as a docker container because certain pre-requisites to the pure installations will not run on that machine. Configurations are pretty straight forward once you understand how docker actually works. There are some other configuration files involved in most cases but largely they are partial replicas of the docker-compose files. FleetDM is really quite impressive the more that I figure out about it. I truly wish I could run a full version instead of having to run this containerized version. There are just some features that are not available in the community version that are included in the professional version.
Now lets move on down the wire. In addition to these two machines are a couple old laptops that also function as servers. 1 of them is a 15 year old Acer (laptop) that also runs Ubuntu. It also runs a containerized version of zeek called network-watcher. Zeek, formerly known as BRO, is basically like a flight recorder for your network. It does not act as a traditional inline firewall to block packets, nor does it strictly yell "Danger!" every time a packet passes by. Instead, it sits passively on a network tap or a SPAN port, watches the traffic, and transforms chaotic raw packet streams into clean, organized, tab-delimited text or JSON log files. In my case these streams are then fed into an LLM for further analysis and then I wrote a python script to interact with the LLM and push the interpretation back out to Loki. Also on this server is an Nginx web server that I use for test driving modifications to this website before I bring them here.
There is also a debian server (old laptop) that sits on the network, and right now its sole purpose is to run the LLM. There are many other uses planned for the LLM, but for right now this is just how I chose to use it. I am thinking that maybe tomorrow morning I will begin trying to set up all of the MDM stuff in FleetDM. But If I am going to do that I may as well integrate some cloud services as well because I will already have to mess with some of it to get MDM working, at least for my phone.
Now lets make this all a little bit more interesting. Every host is connected to a Mesh Overlay VPN [Tailscale]. This DOES in fact pose a potential risk, in that if one host is compromised, it also provides a direct path to each of the others. As a theoretical workaround, the linux hosts all run fail2ban and the windows server runs ipban. The concept is that if brute-force type traffic is detected, it will be dropped at the local firewall. I failed to mention that there is also an instance of Hyper-V running that hosts a Kali VM in bridged mode so that I can actually audit my own set up.
Where things kind of took a turn and had me re-evaluating my momentary choices, was bringing a silicon Mac into the picture. It's just another IP address, Just another Hostname, right? Well, in theory, yes it is right, however it does not speak the same in terms of logs, and Apple is pretty funny about some of their permission stuff... Anyhow it was the last machine that I brought into the mix. Well except for the one that I am doing this on, but this one is a simple integration at this point. I was looking at how to add my phone as well, and that is going to be a whole few days of playing around in cloud services, so if I am going to do it, I may as well wait until I have the money to actually do something with it. This was built and configured in roughly 30 days, and now I spend time working on automation and figuring out policies.
Last night I struggled for a while before I finally gave up and went to bed. I have automated update scripts so that I can update each host from my FleetDM Dashboard. I wanted to implement a policy that made each of them check for and install updates twice a day. BUT apparently I did not want it bad enough, because tonight I am still doing it one host at a time. I just have to do some additional research, that is all.
Additionally I spent some time today tinkering around with some python. It is nothing fancy, but it does save me from having to open that VM whenever I want to see what my surface area looks like. Usually I place the code over on the projects page, but I think this one will be fine right here. This is not designed to take the place of anything. It is just a quick little tool, but nmap at the very least is required.
```python
#!/usr/bin/env python3 reconpipe.py
# Personal recon orchestration pipeline for home-lab use.
# Chains: target expansion -> host discovery -> nmap service/version scan -> targeted enumeration
# (whatweb, nikto, gobuster, smb, dns AXFR, ftp anon).
# Produces summary.json (structured) and summary.md (human-readable).
# Only nmap is mandatory. Missing tools are skipped gracefully.
import argparse
import concurrent.futures
import ipaddress
import json
import shutil
import socket
import subprocess
import sys
import xml.etree.ElementTree as ET
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Set, Tuple
# Dynamic wordlist selection
DEFAULT_WORDLIST = "/usr/share/wordlists/dirb/common.txt"
for path in [
"/usr/share/wordlists/dirb/common.txt",
"/usr/share/wordlists/dirbuster/directory-list-2.3-medium.txt"
]:
if Path(path).exists():
DEFAULT_WORDLIST = path
break
DEFAULT_OUTDIR = "recon_results"
DEFAULT_WORKERS = 5
TOOL_TIMEOUT = 600 # seconds
WEB_SERVICES = {"http", "https", "http-proxy", "http-alt"}
SMB_PORTS = {139, 445}
FTP_PORTS = {21}
DNS_PORTS = {53}
def log(msg: str) -> None:
print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)
def have(tool: str) -> bool:
return shutil.which(tool) is not None
def run(cmd: List[str], timeout: int = TOOL_TIMEOUT) -> Tuple[int, str, str]:
"""Run command and return (returncode, stdout, stderr)."""
try:
p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
return p.returncode, p.stdout, p.stderr
except subprocess.TimeoutExpired:
return -1, "", f"TIMEOUT after {timeout}s"
except FileNotFoundError:
return -1, "", f"binary not found: {cmd[0]}"
def resolve(host: str) -> str | None:
try:
return socket.gethostbyname(host)
except socket.gaierror:
return None
# ========================== SCOPE ==========================
def load_scope(path: Path) -> Tuple[List[ipaddress.IPv4Network | ipaddress.IPv6Network], List[str]]:
nets, hosts = [], []
for line in path.read_text().splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
try:
nets.append(ipaddress.ip_network(line, strict=False))
except ValueError:
hosts.append(line.lower())
return nets, hosts
def in_scope(target: str, nets, hosts: List[str]) -> bool:
if target.lower() in hosts:
return True
try:
ip = ipaddress.ip_address(target)
return any(ip in n for n in nets)
except ValueError:
pass
resolved = resolve(target)
if resolved:
return any(ipaddress.ip_address(resolved) in n for n in nets)
return False
# ========================== TARGETS ==========================
def expand_targets(target_arg: str) -> List[str]:
p = Path(target_arg)
if p.is_file():
return [l.strip() for l in p.read_text().splitlines()
if l.strip() and not l.startswith("#")]
return [target_arg]
# ========================== DISCOVERY ==========================
def parse_discovery_xml(xml_path: Path) -> Set[str]:
live = set()
if not xml_path.exists():
return live
tree = ET.parse(xml_path)
for host in tree.findall(".//host"):
if host.find(".//status[@state='up']") is not None:
addr = host.find(".//address")
if addr is not None:
live.add(addr.get("addr"))
return live
def discover_hosts(targets: List[str], outdir: Path) -> List[str]:
cidrs = [t for t in targets if "/" in t]
singles = [t for t in targets if "/" not in t]
live = set(singles)
if cidrs:
xml_path = outdir / "discovery.xml"
log(f"Host discovery on {', '.join(cidrs)} ...")
rc, _, err = run(["nmap", "-sn", "-oX", str(xml_path), *cidrs], timeout=300)
if rc != 0 and err:
log(f"discovery warning: {err.strip()[:200]}")
live |= parse_discovery_xml(xml_path)
return sorted(live)
# ========================== PORT SCAN ==========================
def parse_service_scan_xml(xml_path: Path) -> Dict:
results = {}
if not xml_path.exists():
return results
tree = ET.parse(xml_path)
for host in tree.findall(".//host"):
addr = host.find(".//address[@addrtype='ipv4']")
if addr is None:
continue
ip = addr.get("addr")
hostname = None
hn = host.find(".//hostname")
if hn is not None:
hostname = hn.get("name")
ports = []
for port_el in host.findall(".//port"):
state = port_el.find("state")
if state is None or state.get("state") != "open":
continue
portid = int(port_el.get("portid"))
proto = port_el.get("protocol", "tcp")
svc_el = port_el.find("service")
service = svc_el.get("name", "unknown") if svc_el is not None else "unknown"
product = svc_el.get("product", "") if svc_el is not None else ""
version = svc_el.get("version", "") if svc_el is not None else ""
banner = f"{product} {version}".strip()
notes = [
f'{s.get("id")}: {(s.get("output") or "").replace("\\n", " ")[:200]}'
for s in port_el.findall("script")
]
ports.append({
"port": portid,
"proto": proto,
"service": service,
"banner": banner,
"nse_notes": notes,
})
results[ip] = {"hostname": hostname, "ports": ports}
return results
def scan_ports(hosts: List[str], outdir: Path, full_ports: bool, workers: int):
xml_path = outdir / "scan.xml"
port_flag = ["-p-"] if full_ports else ["--top-ports", "1000"]
log(f"Service scan on {len(hosts)} host(s) this may take a while...")
cmd = ["nmap", "-T4", "-Pn", "-sV", "-sC", *port_flag,
"--min-parallelism", str(workers), "-oX", str(xml_path), *hosts]
rc, _, err = run(cmd, timeout=3600)
if rc != 0 and err:
log(f"nmap scan warning: {err.strip()[:300]}")
return parse_service_scan_xml(xml_path)
# ========================== ENUMERATION ==========================
def save_output(host_dir: Path, filename: str, content: str) -> None:
(host_dir / filename).write_text(content)
def enum_web(host: str, port: int, scheme: str, host_dir: Path, wordlist: str) -> List[str]:
base = f"{scheme}://{host}:{port}"
findings = []
# WhatWeb
if have("whatweb"):
rc, out, err = run(["whatweb", "--color=never", base])
save_output(host_dir, f"whatweb_{port}.txt", out + err)
if scheme == "http" and (rc != 0 or "400" in out or "Control Channel" in err):
log(f" [!] Retrying {port} with HTTPS...")
base = f"https://{host}:{port}"
rc, out, err = run(["whatweb", "--color=never", base])
save_output(host_dir, f"whatweb_{port}.txt", out + err)
if out.strip():
findings.append(f"whatweb: {out.strip().splitlines()[-1][:300]}")
# Nikto
if have("nikto"):
out_file = host_dir / f"nikto_{port}.txt"
run(["nikto", "-h", base, "-ask", "no", "-o", str(out_file), "-Format", "txt"], timeout=900)
if out_file.stat().st_size > 0:
findings.append(f"nikto: see {out_file.name}")
# Gobuster
if have("gobuster") and Path(wordlist).exists():
out_file = host_dir / f"gobuster_{port}.txt"
run(["gobuster", "dir", "-u", base, "-w", wordlist, "-q", "-k",
"-o", str(out_file)], timeout=900)
if out_file.exists() and out_file.stat().st_size > 0:
hits = sum(1 for line in out_file.read_text().splitlines()
if line.strip() and not line.startswith('#'))
findings.append(f"gobuster: {hits} path(s) found, see {out_file.name}")
return findings
def enum_smb(host: str, host_dir: Path) -> List[str]:
findings = []
if have("enum4linux-ng"):
out_file = host_dir / "enum4linux-ng.txt"
_, out, err = run(["enum4linux-ng", "-A", host], timeout=600)
save_output(host_dir, "enum4linux-ng.txt", out + err)
findings.append(f"enum4linux-ng: see {out_file.name}")
elif have("smbclient"):
_, out, err = run(["smbclient", "-L", f"//{host}", "-N"], timeout=60)
save_output(host_dir, "smbclient.txt", out + err)
if "Sharename" in out:
findings.append("smbclient: anonymous share listing succeeded")
return findings
def enum_dns_axfr(host: str, host_dir: Path, domain_hint: str | None) -> List[str]:
if not (have("dig") and domain_hint):
return []
_, out, _ = run(["dig", f"@{host}", domain_hint, "AXFR"], timeout=30)
save_output(host_dir, "dns_axfr.txt", out)
if "Transfer failed" not in out and "communications error" not in out and out.count("\n") > 10:
return ["dns: zone transfer (AXFR) may have succeeded review dns_axfr.txt"]
return []
def check_ftp_anon(host: str, port: int, host_dir: Path) -> List[str]:
_, out, _ = run(["nmap", "-p", str(port), "--script", "ftp-anon", host], timeout=60)
save_output(host_dir, f"ftp_anon_{port}.txt", out)
if "Anonymous FTP login allowed" in out:
return ["ftp: anonymous login allowed"]
return []
def enumerate_host(ip: str, hostdata: Dict, outdir: Path, wordlist: str, domain_hint: str | None):
host_dir = outdir / ip.replace(":", "_")
host_dir.mkdir(parents=True, exist_ok=True)
findings = []
for p in hostdata.get("ports", []):
port = p["port"]
svc = p["service"].lower()
if svc in WEB_SERVICES or "http" in svc:
scheme = "https" if ("ssl" in svc or "https" in svc or port == 443) else "http"
findings.extend([f"{port}/tcp - {f}" for f in enum_web(ip, port, scheme, host_dir, wordlist)])
elif port in SMB_PORTS:
findings.extend([f"{port}/tcp - {f}" for f in enum_smb(ip, host_dir)])
elif port in DNS_PORTS:
findings.extend([f"{port}/tcp - {f}" for f in enum_dns_axfr(ip, host_dir, domain_hint)])
elif port in FTP_PORTS:
findings.extend([f"{port}/tcp - {f}" for f in check_ftp_anon(ip, port, host_dir)])
return ip, findings
# ========================== REPORTING ==========================
def write_reports(outdir: Path, scan_results: Dict, findings_by_host: Dict, run_id: str, targets: List[str]):
summary = {
"run_id": run_id,
"targets_requested": targets,
"hosts": {}
}
for ip, data in scan_results.items():
summary["hosts"][ip] = {
"hostname": data.get("hostname"),
"ports": data["ports"],
"findings": findings_by_host.get(ip, []),
}
(outdir / "summary.json").write_text(json.dumps(summary, indent=2))
# Markdown report
lines = [f"# Recon run {run_id}\n"]
flagged = [(ip, f) for ip, fs in findings_by_host.items() for f in fs]
if flagged:
lines.append("## Notable findings")
for ip, f in flagged:
lines.append(f"- **{ip}** {f}")
lines.append("")
for ip, data in scan_results.items():
hn = f" ({data['hostname']})" if data.get("hostname") else ""
lines.append(f"## {ip}{hn}")
if not data["ports"]:
lines.append("_no open ports found_")
continue
lines.append("| Port | Service | Banner |")
lines.append("|---|---|---|")
for p in data["ports"]:
lines.append(f"| {p['port']}/{p['proto']} | {p['service']} | {p['banner']} |")
lines.append("")
(outdir / "summary.md").write_text("\n".join(lines))
log(f"Reports written {outdir}/summary.md and summary.json")
# ========================== MAIN ==========================
def main():
ap = argparse.ArgumentParser(description="Personal recon orchestration pipeline.")
ap.add_argument("-t", "--target", required=True, help="single IP/host, CIDR, or file with targets")
ap.add_argument("--scope", type=Path, help="scope allowlist file")
ap.add_argument("-o", "--outdir", default=DEFAULT_OUTDIR, type=Path)
ap.add_argument("--wordlist", default=DEFAULT_WORDLIST, type=Path)
ap.add_argument("--workers", type=int, default=DEFAULT_WORKERS)
ap.add_argument("--full-ports", action="store_true")
ap.add_argument("--domain", help="Domain for DNS AXFR attempts (e.g. home.lab)")
ap.add_argument("-y", "--yes", action="store_true", help="skip confirmation")
args = ap.parse_args()
if not have("nmap"):
sys.exit("Error: nmap is required but not found on PATH.")
targets = expand_targets(args.target)
if args.scope:
nets, hosts = load_scope(args.scope)
bad = [t for t in targets if not in_scope(t, nets, hosts)]
if bad:
sys.exit(f"Target(s) outside scope: {bad}")
run_id = datetime.now().strftime("%Y%m%d_%H%M%S")
outdir = args.outdir / run_id
outdir.mkdir(parents=True, exist_ok=True)
log(f"Targets: {targets}")
if not args.yes:
if input("Proceed with scan? [y/N] ").strip().lower() != "y":
sys.exit("Aborted.")
live_hosts = discover_hosts(targets, outdir)
if args.scope:
live_hosts = [h for h in live_hosts if in_scope(h, nets, hosts)]
if not live_hosts:
sys.exit("No live in-scope hosts found.")
log(f"Live hosts: {live_hosts}")
scan_results = scan_ports(live_hosts, outdir, args.full_ports, args.workers)
log("Starting targeted enumeration...")
findings_by_host = {}
with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as ex:
futures = [
ex.submit(enumerate_host, ip, data, outdir, str(args.wordlist), args.domain)
for ip, data in scan_results.items()
]
for fut in concurrent.futures.as_completed(futures):
ip, findings = fut.result()
findings_by_host[ip] = findings
log(f"Enumeration done: {ip} ({len(findings)} findings)")
write_reports(outdir, scan_results, findings_by_host, run_id, targets)
if __name__ == "__main__":
main()