Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/nssec/cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -191,9 +191,11 @@ def init(config_dir):
# ─── REGISTER SUB-COMMAND GROUPS ───

from nssec.cli.audit import audit # noqa: E402
from nssec.cli.mtls_commands import mtls # noqa: E402
from nssec.cli.waf_commands import waf # noqa: E402

cli.add_command(audit)
cli.add_command(mtls)
cli.add_command(waf)


Expand Down
175 changes: 175 additions & 0 deletions src/nssec/cli/mtls_commands.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,175 @@
"""mTLS management CLI commands for nssec."""

import click

from nssec.cli import console


@click.group()
def mtls():
"""mTLS device provisioning management commands."""
pass


@mtls.group("nodeping", invoke_without_command=True)
@click.pass_context
def mtls_nodeping(ctx):
"""Manage NodePing monitoring IPs in mTLS config."""
if ctx.invoked_subcommand is None:
ctx.invoke(nodeping_show)


@mtls_nodeping.command("show")
def nodeping_show():
"""Show current NodePing IPs in ndp_mtls.conf."""
from nssec.modules.mtls import get_current_nodeping_ips
from nssec.modules.mtls.config import NDP_MTLS_CONF
from nssec.modules.mtls.utils import file_exists

if not file_exists(NDP_MTLS_CONF):
console.print(f"[yellow]mTLS config not found:[/yellow] {NDP_MTLS_CONF}")
console.print("[dim]Is mTLSProtect installed?[/dim]")
return

ips = get_current_nodeping_ips()
if not ips:
console.print("[dim]No NodePing IPs currently configured.[/dim]")
console.print("\nTo add NodePing IPs, run:")
console.print(" [cyan]sudo nssec mtls nodeping update[/cyan]")
return

_display_ip_list(ips, "NodePing IPs")


@mtls_nodeping.command("fetch")
def nodeping_fetch():
"""Fetch and display NodePing IPs (dry run, no changes)."""
from nssec.modules.mtls.utils import fetch_nodeping_ips

console.print("[bold]Fetching NodePing IPs...[/bold]")
ips, error = fetch_nodeping_ips()

if error:
console.print(f"[red]Error:[/red] {error}")
raise SystemExit(1)

console.print(f"\n[green]Fetched {len(ips)} IPs from NodePing[/green]\n")
_display_ip_list(ips, "Available IPs")

console.print("\n[dim]This was a dry run. To apply, run:[/dim]")
console.print(" [cyan]sudo nssec mtls nodeping update[/cyan]")


def _require_root(command_name: str) -> None:
"""Exit with error if not running as root."""
from nssec.core.ssh import is_root

if not is_root():
console.print(
f"[red]Error: Must run as root (sudo nssec mtls nodeping {command_name})[/red]"
)
raise SystemExit(1)


def _validate_and_reload(yes: bool) -> None:
"""Validate Apache config and optionally reload."""
from nssec.modules.mtls import reload_apache, rollback, validate_apache_config
from nssec.modules.mtls.config import NDP_MTLS_CONF

val_result = validate_apache_config()
if not val_result.success:
console.print(f" [red]Error:[/red] {val_result.error}")
console.print("[yellow]Rolling back changes...[/yellow]")
rollback(NDP_MTLS_CONF)
raise SystemExit(1)
console.print(f" [green]Done:[/green] {val_result.message}")

console.print()
if yes or click.confirm("Reload Apache to apply changes?"):
reload_result = reload_apache()
if reload_result.success:
console.print(f" [green]Done:[/green] {reload_result.message}")
else:
console.print(f" [red]Error:[/red] {reload_result.error}")
raise SystemExit(1)
else:
console.print("[yellow]Skipped Apache reload. Run manually:[/yellow]")
console.print(" [cyan]sudo systemctl reload apache2[/cyan]")


@mtls_nodeping.command("update")
@click.option("--yes", "-y", is_flag=True, help="Skip confirmation prompts")
@click.option("--dry-run", is_flag=True, help="Show what would be done without changes")
def nodeping_update(yes, dry_run):
"""Fetch NodePing IPs and update ndp_mtls.conf."""
from nssec.modules.mtls import update_nodeping_ips

_require_root("update")
console.print("[bold]Updating NodePing IPs...[/bold]")

result = update_nodeping_ips(dry_run=dry_run)

if result.skipped:
console.print(f"[green]{result.message}[/green]")
return

if not result.success:
console.print(f"[red]Error:[/red] {result.error}")
raise SystemExit(1)

console.print(f" [green]Done:[/green] {result.message}")

if dry_run:
console.print("\n[yellow]Dry run - no changes made.[/yellow]")
return

_validate_and_reload(yes)


@mtls_nodeping.command("remove")
@click.option("--yes", "-y", is_flag=True, help="Skip confirmation")
def nodeping_remove(yes):
"""Remove NodePing IPs section from ndp_mtls.conf."""
from nssec.modules.mtls import remove_nodeping_ips

_require_root("remove")

console.print(
"[bold yellow]Warning:[/bold yellow] This will remove all NodePing IPs from mTLS config."
)
console.print()

if not yes and not click.confirm("Remove NodePing IPs section?"):
console.print("[yellow]Aborted.[/yellow]")
return

result = remove_nodeping_ips()

if result.skipped:
console.print(f"[dim]{result.message}[/dim]")
return

if not result.success:
console.print(f"[red]Error:[/red] {result.error}")
raise SystemExit(1)

console.print(f" [green]Done:[/green] {result.message}")
_validate_and_reload(yes)


def _display_ip_list(ips: list[str], title: str) -> None:
"""Display a list of IPs grouped by version."""
ipv4 = [ip for ip in ips if ":" not in ip]
ipv6 = [ip for ip in ips if ":" in ip]

console.print(f"[bold]{title}[/bold] ({len(ips)} total)\n")

if ipv4:
console.print(f"[cyan]IPv4[/cyan] ({len(ipv4)}):")
for ip in sorted(ipv4):
console.print(f" {ip}")

if ipv6:
console.print(f"\n[cyan]IPv6[/cyan] ({len(ipv6)}):")
for ip in sorted(ipv6):
console.print(f" {ip}")
95 changes: 94 additions & 1 deletion src/nssec/cli/waf_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,8 @@ def _display_install_plan(pf, mode, skip_evasive):
table.add_row("security2.conf", sec2_state, sec2_action)

if not skip_evasive:
table.add_row("mod_evasive", "", "install if missing")
evasive_action = "install + enable" if mode == "On" else "install config (disabled in DetectionOnly)"
table.add_row("mod_evasive", "", evasive_action)

console.print(table)

Expand Down Expand Up @@ -133,6 +134,12 @@ def _build_status_table(status):
else:
table.add_row("OWASP CRS", "[red]not installed[/red]")

if status.evasive_installed:
evasive_state = "[green]enabled[/green]" if status.evasive_enabled else "[yellow]disabled[/yellow]"
table.add_row("mod_evasive", evasive_state)
else:
table.add_row("mod_evasive", "[dim]not installed[/dim]")

table.add_row("NS exclusions", _yn(status.exclusions_present, "yellow"))
table.add_row("Audit log", _yn(status.audit_log_exists, "dim"))
return table
Expand Down Expand Up @@ -218,6 +225,9 @@ def waf_enable(yes):
"[bold yellow]Warning:[/bold yellow] Switching to blocking mode "
"will actively reject requests that match ModSecurity rules."
)
console.print(
"This will also [bold]enable mod_evasive[/bold] (HTTP flood protection)."
)
console.print(
"Ensure you have reviewed "
"[cyan]/var/log/apache2/modsec_audit.log[/cyan] for false positives."
Expand All @@ -236,6 +246,89 @@ def waf_enable(yes):
raise SystemExit(1)


@waf.command("disable")
@click.option("--yes", "-y", is_flag=True, help="Skip confirmation")
def waf_disable(yes):
"""Switch ModSecurity to DetectionOnly mode (logs but does not block)."""
from nssec.modules.waf import ModSecurityInstaller

installer = ModSecurityInstaller()
pf = installer.preflight()
_require_root_and_modsec(pf, "sudo nssec waf disable")

if pf.modsec_mode and pf.modsec_mode.lower() == "detectiononly":
console.print("[green]ModSecurity is already in DetectionOnly mode.[/green]")
return

console.print(
"Switching to [cyan]DetectionOnly[/cyan] mode. "
"ModSecurity will log violations but not block requests."
)
console.print(
"This will also [bold]disable mod_evasive[/bold] (HTTP flood protection)."
)
console.print()

if not yes and not click.confirm("Switch SecRuleEngine to DetectionOnly?"):
console.print("[yellow]Aborted.[/yellow]")
return

result = installer.set_mode("DetectionOnly")
if result.success:
console.print(f"[green]{result.message}[/green]")
else:
console.print(f"[red]Error: {result.error}[/red]")
raise SystemExit(1)


@waf.command("remove")
@click.option("--yes", "-y", is_flag=True, help="Skip confirmation")
def waf_remove(yes):
"""Disable the ModSecurity Apache module entirely.

This disables the security2 module in Apache, effectively turning off
the WAF completely. Use 'nssec waf init' to re-enable.
"""
from nssec.modules.waf import ModSecurityInstaller
from nssec.modules.waf.utils import run_cmd, file_exists
from nssec.modules.waf.config import SECURITY2_LOAD
from nssec.core.ssh import is_root

if not is_root():
console.print("[red]Error: Must run as root (sudo nssec waf remove)[/red]")
raise SystemExit(1)

if not file_exists(SECURITY2_LOAD):
console.print("[green]ModSecurity module is already disabled.[/green]")
return

console.print(
"[bold yellow]Warning:[/bold yellow] This will completely disable "
"the ModSecurity WAF module."
)
console.print()

if not yes and not click.confirm("Disable ModSecurity module?"):
console.print("[yellow]Aborted.[/yellow]")
return

_, stderr, rc = run_cmd(["a2dismod", "security2"])
if rc != 0:
console.print(f"[red]Error:[/red] Failed to disable module: {stderr}")
raise SystemExit(1)
console.print("[green]Done:[/green] Disabled security2 module")

_, stderr, rc = run_cmd(["systemctl", "reload", "apache2"])
if rc != 0:
console.print(f"[red]Error:[/red] Apache reload failed: {stderr}")
raise SystemExit(1)
console.print("[green]Done:[/green] Apache reloaded")

console.print()
console.print("ModSecurity is now disabled. To re-enable, run:")
console.print(" [cyan]sudo nssec waf init[/cyan]")


@waf.command("update-exclusions")
@click.option("--yes", "-y", is_flag=True, help="Skip confirmation")
@click.option(
Expand Down
Loading