diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..f2361ea --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,57 @@ +name: CI + +on: + push: + branches: [master, "cursor/**"] + pull_request: + branches: [master] + +jobs: + check: + runs-on: ubuntu-latest + services: + mysql: + image: mysql:8.0 + env: + MYSQL_ROOT_PASSWORD: root + MYSQL_DATABASE: asprom + MYSQL_USER: asprom + MYSQL_PASSWORD: asprom + ports: + - 3306:3306 + options: >- + --health-cmd="mysqladmin ping -h localhost" + --health-interval=10s + --health-timeout=5s + --health-retries=5 + env: + ASPROM_TEST_DB_HOST: 127.0.0.1 + ASPROM_TEST_DB_PORT: 3306 + ASPROM_TEST_DB_USER: asprom + ASPROM_TEST_DB_PASSWORD: asprom + ASPROM_TEST_DB_NAME: asprom + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install system dependencies + run: | + sudo apt-get update + sudo apt-get install -y default-libmysqlclient-dev build-essential pkg-config + + - name: Install Python dependencies + run: | + pip install -r requirements.txt + pip install pytest pytest-cov "testcontainers[mysql]" ruff mypy + + - name: Run check script + env: + ASPROM_COV_FAIL_UNDER: "70" + ASPROM_RUN_MYPY: "1" + run: ./scripts/check.sh + + - name: Docker build smoke test + run: docker build . diff --git a/.gitignore b/.gitignore index 2837890..069e086 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,7 @@ .env *.swp *.bak +__pycache__/ +*.pyc +.coverage +.pytest_cache/ diff --git a/README.md b/README.md index ace42b1..99e9d02 100644 --- a/README.md +++ b/README.md @@ -43,3 +43,72 @@ Access metrics about open ports and baseline deviations at: ### Nagios Integration Use `aspromNagiosCheck.py` as a standard Nagios plugin to receive active alerts. The plugin will return CRITICAL status when unauthorized services are detected. +## Development + +[![CI](https://github.com/daimoniac/asprom/actions/workflows/ci.yml/badge.svg)](https://github.com/daimoniac/asprom/actions/workflows/ci.yml) + +Run the full project check before every commit: + +```bash +ASPROM_COV_FAIL_UNDER=70 ./scripts/check.sh +``` + +Optional git pre-commit hook: + +```bash +ln -sf ../../scripts/check.sh .git/hooks/pre-commit +``` + +Install Python dependencies (use a venv on Debian/Ubuntu — system Python is externally managed): + +```bash +# one-time system packages (Debian/Ubuntu) +sudo apt install -y python3-venv default-libmysqlclient-dev pkg-config build-essential nmap + +# create venv and install deps +./scripts/setup-venv.sh +source venv/bin/activate +``` + +Or manually: + +```bash +python3 -m venv venv +source venv/bin/activate # not: venv/bin/activate +pip install -r requirements.txt pytest pytest-cov "testcontainers[mysql]" ruff mypy sqlalchemy +``` + +### Database migrations + +- **Fresh Docker installs:** schema applied via `db/ddl.sql` on first MySQL container start. +- **Existing installs:** run `alembic stamp 001` then `alembic upgrade head`. +- **Future schema changes:** add Alembic revisions only. + +Set `ASPROM_RUN_MIGRATIONS=1` in the asprom container to run `alembic upgrade head` on startup. + +### Integration tests + +Tests use `ASPROM_TEST_DB_*` environment variables (set automatically in CI via GitHub Actions MySQL service). Locally, a MySQL instance on `127.0.0.1` with database `asprom_test` is used by default. + +### Local dev against production Kubernetes DB + +Run the legacy Bottle GUI locally while port-forwarding the production MySQL service from Kubernetes: + +```bash +./scripts/setup-venv.sh +source venv/bin/activate +./scripts/dev-prod.sh +``` + +The script discovers the MySQL service name via `kubectl` (default context `internal1`, namespace `asprom`), port-forwards it to `127.0.0.1:3307`, and starts `aspromGUI.py` on [http://127.0.0.1:8080](http://127.0.0.1:8080). + +Override discovery if needed: + +```bash +KUBE_CONTEXT=internal1 KUBE_NAMESPACE=asprom MYSQL_SERVICE=mysql ./scripts/dev-prod.sh +``` + +Set `ASPROM_DB_PASSWORD` if the script cannot read credentials from a Kubernetes secret. + +**Warning:** this connects to production data. Scans and baseline changes affect live systems. + diff --git a/__init__.py b/__init__.py deleted file mode 100644 index f649d22..0000000 --- a/__init__.py +++ /dev/null @@ -1,4 +0,0 @@ -# __init__.py file - -import pymysql -pymysql.install_as_MySQLdb() diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..807ded2 --- /dev/null +++ b/alembic.ini @@ -0,0 +1,149 @@ +# A generic, single database configuration. + +[alembic] +# path to migration scripts. +# this is typically a path given in POSIX (e.g. forward slashes) +# format, relative to the token %(here)s which refers to the location of this +# ini file +script_location = %(here)s/alembic + +# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s +# Uncomment the line below if you want the files to be prepended with date and time +# see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file +# for all available tokens +# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s +# Or organize into date-based subdirectories (requires recursive_version_locations = true) +# file_template = %%(year)d/%%(month).2d/%%(day).2d_%%(hour).2d%%(minute).2d_%%(second).2d_%%(rev)s_%%(slug)s + +# sys.path path, will be prepended to sys.path if present. +# defaults to the current working directory. for multiple paths, the path separator +# is defined by "path_separator" below. +prepend_sys_path = . + + +# timezone to use when rendering the date within the migration file +# as well as the filename. +# If specified, requires the tzdata library which can be installed by adding +# `alembic[tz]` to the pip requirements. +# string value is passed to ZoneInfo() +# leave blank for localtime +# timezone = + +# max length of characters to apply to the "slug" field +# truncate_slug_length = 40 + +# set to 'true' to run the environment during +# the 'revision' command, regardless of autogenerate +# revision_environment = false + +# set to 'true' to allow .pyc and .pyo files without +# a source .py file to be detected as revisions in the +# versions/ directory +# sourceless = false + +# version location specification; This defaults +# to /versions. When using multiple version +# directories, initial revisions must be specified with --version-path. +# The path separator used here should be the separator specified by "path_separator" +# below. +# version_locations = %(here)s/bar:%(here)s/bat:%(here)s/alembic/versions + +# path_separator; This indicates what character is used to split lists of file +# paths, including version_locations and prepend_sys_path within configparser +# files such as alembic.ini. +# The default rendered in new alembic.ini files is "os", which uses os.pathsep +# to provide os-dependent path splitting. +# +# Note that in order to support legacy alembic.ini files, this default does NOT +# take place if path_separator is not present in alembic.ini. If this +# option is omitted entirely, fallback logic is as follows: +# +# 1. Parsing of the version_locations option falls back to using the legacy +# "version_path_separator" key, which if absent then falls back to the legacy +# behavior of splitting on spaces and/or commas. +# 2. Parsing of the prepend_sys_path option falls back to the legacy +# behavior of splitting on spaces, commas, or colons. +# +# Valid values for path_separator are: +# +# path_separator = : +# path_separator = ; +# path_separator = space +# path_separator = newline +# +# Use os.pathsep. Default configuration used for new projects. +path_separator = os + +# set to 'true' to search source files recursively +# in each "version_locations" directory +# new in Alembic version 1.10 +# recursive_version_locations = false + +# the output encoding used when revision files +# are written from script.py.mako +# output_encoding = utf-8 + +# database URL. This is consumed by the user-maintained env.py script only. +# other means of configuring database URLs may be customized within the env.py +# file. +sqlalchemy.url = driver://user:pass@localhost/dbname + + +[post_write_hooks] +# post_write_hooks defines scripts or Python functions that are run +# on newly generated revision scripts. See the documentation for further +# detail and examples + +# format using "black" - use the console_scripts runner, against the "black" entrypoint +# hooks = black +# black.type = console_scripts +# black.entrypoint = black +# black.options = -l 79 REVISION_SCRIPT_FILENAME + +# lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module +# hooks = ruff +# ruff.type = module +# ruff.module = ruff +# ruff.options = check --fix REVISION_SCRIPT_FILENAME + +# Alternatively, use the exec runner to execute a binary found on your PATH +# hooks = ruff +# ruff.type = exec +# ruff.executable = ruff +# ruff.options = check --fix REVISION_SCRIPT_FILENAME + +# Logging configuration. This is also consumed by the user-maintained +# env.py script only. +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARNING +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARNING +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/alembic/README b/alembic/README new file mode 100644 index 0000000..98e4f9c --- /dev/null +++ b/alembic/README @@ -0,0 +1 @@ +Generic single-database configuration. \ No newline at end of file diff --git a/alembic/env.py b/alembic/env.py new file mode 100644 index 0000000..eae9358 --- /dev/null +++ b/alembic/env.py @@ -0,0 +1,58 @@ +import os +from logging.config import fileConfig + +from sqlalchemy import engine_from_config, pool + +from alembic import context + +config = context.config + +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +target_metadata = None + + +def _database_url() -> str: + if url := os.environ.get("ASPROM_DB_URL"): + return url + host = os.environ.get("ASPROM_TEST_DB_HOST", os.environ.get("MYSQL_HOST", "127.0.0.1")) + port = os.environ.get("ASPROM_TEST_DB_PORT", os.environ.get("MYSQL_PORT", "3306")) + user = os.environ.get("ASPROM_TEST_DB_USER", os.environ.get("MYSQL_USER", "asprom")) + password = os.environ.get("ASPROM_TEST_DB_PASSWORD", os.environ.get("MYSQL_PASSWORD", "asprom")) + database = os.environ.get("ASPROM_TEST_DB_NAME", os.environ.get("MYSQL_DATABASE", "asprom")) + return f"mysql+mysqldb://{user}:{password}@{host}:{port}/{database}" + + +def run_migrations_offline() -> None: + context.configure( + url=_database_url(), + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + ) + + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online() -> None: + configuration = config.get_section(config.config_ini_section, {}) + configuration["sqlalchemy.url"] = _database_url() + connectable = engine_from_config( + configuration, + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + + with connectable.connect() as connection: + context.configure(connection=connection, target_metadata=target_metadata) + + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/alembic/script.py.mako b/alembic/script.py.mako new file mode 100644 index 0000000..1101630 --- /dev/null +++ b/alembic/script.py.mako @@ -0,0 +1,28 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +# revision identifiers, used by Alembic. +revision: str = ${repr(up_revision)} +down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)} +branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)} +depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)} + + +def upgrade() -> None: + """Upgrade schema.""" + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + """Downgrade schema.""" + ${downgrades if downgrades else "pass"} diff --git a/alembic/versions/001_baseline.py b/alembic/versions/001_baseline.py new file mode 100644 index 0000000..68571d0 --- /dev/null +++ b/alembic/versions/001_baseline.py @@ -0,0 +1,14 @@ +"""Baseline schema — existing installs should stamp at this revision.""" + +revision = "001" +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade() -> None: + pass + + +def downgrade() -> None: + pass diff --git a/alembic/versions/002_rename_unique_keys.py b/alembic/versions/002_rename_unique_keys.py new file mode 100644 index 0000000..58020da --- /dev/null +++ b/alembic/versions/002_rename_unique_keys.py @@ -0,0 +1,18 @@ +"""Rename German unique key names to English identifiers.""" + +from alembic import op + +revision = "002" +down_revision = "001" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.execute("ALTER TABLE services RENAME INDEX `Schlüssel 2` TO uk_port_machine") + op.execute("ALTER TABLE machines RENAME INDEX `Schlüssel 2` TO uk_ip") + + +def downgrade() -> None: + op.execute("ALTER TABLE services RENAME INDEX uk_port_machine TO `Schlüssel 2`") + op.execute("ALTER TABLE machines RENAME INDEX uk_ip TO `Schlüssel 2`") diff --git a/aspromGUI.py b/aspromGUI.py index 20d71b3..966ac65 100644 --- a/aspromGUI.py +++ b/aspromGUI.py @@ -1,4 +1,4 @@ -''' +""" Created on Oct 19, 2014 @author stefankn @@ -6,16 +6,20 @@ Main Script for the asprom GUI. This script presents a webserver socket to which client browsers can connect to. Also, it orchestrates URL calls between the model, view and controller classes. -''' -from bottle import (route, run, static_file, abort, redirect, template, - post, request, hook, response) -from inc.asprom import (AspromModel, AspromScheduleModel, Controller, Machine, initDB, - closeDB, Cfg) +""" + +from bottle import abort, hook, post, redirect, request, response, route, run, static_file, template + +from inc.asprom import AspromModel, AspromScheduleModel, Cfg, Controller, Machine, closeDB, initDB +from inc.logging import configure_logging, get_logger + +configure_logging() +logger = get_logger(__name__) # Variable definitions ## relative path to static files -sr = 'static/' +sr = "static/" ## main model M = None @@ -30,19 +34,19 @@ # -@hook('before_request') +@hook("before_request") def before_request(): - ''' + """ before each dynamic request, create DB connection and Model instances. - ''' + """ username = None try: username = request.get_header("X-Forwarded-User", request.auth[0]) - except: + except (TypeError, AttributeError, KeyError): pass p = request.path - if p.startswith('/' + sr): + if p.startswith("/" + sr): return global M, SM try: @@ -52,103 +56,119 @@ def before_request(): except: raise + # # routes # -@route('/') +@route("/") def serve_homepage(): - ''' + """ HTTP Redirect to http:///alerts-exposed. - ''' - redirect('/alerts-exposed') + """ + redirect("/alerts-exposed") + # main views -@route('/alerts-exposed') +@route("/alerts-exposed") def serve_alertsexposed(): - ''' + """ Presents view: http:///alerts-exposed. - ''' - return template('views/alerts-exposed') + """ + return template("views/alerts-exposed") -@route('/alerts-closed') +@route("/alerts-closed") def serve_alertsclosed(): - ''' + """ Presents view: http:///alerts-closed. - ''' - return template('views/alerts-closed') + """ + return template("views/alerts-closed") -@route('/baseline') +@route("/baseline") def serve_baseline(): - ''' + """ Presents view: http:///baseline. - ''' - return template('views/baseline') + """ + return template("views/baseline") -@route('/posture') +@route("/posture") def serve_forensic(): - ''' + """ Presents view: http:///posture. - ''' - return template('views/posture') + """ + return template("views/posture") -@route('/schedule') +@route("/schedule") def serve_schedule(): - ''' + """ Presents view: http:///schedule. - ''' - return template('views/schedule') + """ + return template("views/schedule") -@route('/log') +@route("/log") def serve_log(): - ''' + """ Presents view: http:///log. - ''' + """ return M.getLastLog(10) # dialog views -@route('/dia/editjob/') +@route( + "/dia/editjob/" +) def serve_editjob_view(jobid): - ''' + """ Presents view: http:///dia/editjob. This is meant to be used as a dialog popup in the schedule view. On this dialog, the parameters of an existing job can be edited. @param jobid the job ID to be edited. - ''' + """ j = SM.getScheduleEntryByID(jobid) - return template('views/editjob', jobid=jobid, initial=j['when'], iprange=j - ['iprange'], portrange=j['ports'], extraparams=j['params']) + return template( + "views/editjob", + jobid=jobid, + initial=j["when"], + iprange=j["iprange"], + portrange=j["ports"], + extraparams=j["params"], + ) -@route('/dia/addjob') +@route("/dia/addjob") def serve_addjob_view(): - ''' + """ Presents view: http:///dia/addjob. This is meant to be used as a dialog popup in the schedule view. On this dialog, the parameters of a new job can be entered. - ''' + """ from uuid import uuid4 - return template('views/addjob', jobid=str(uuid4()), initial='0 1 * * *', - iprange='192.168.0.0/24', portrange='0-1024', - extraparams='-sV') + + return template( + "views/addjob", + jobid=str(uuid4()), + initial="0 1 * * *", + iprange="192.168.0.0/24", + portrange="0-1024", + extraparams="-sV", + ) # JSON Views -@route('/json/') +@route("/json/") def returnjson(filename): - ''' + """ Presents all json views: http:///json/*. These are used by the tables embedded in the main html views. The data is aquired using ajax calls. @@ -157,16 +177,16 @@ def returnjson(filename): @param filename the json view to be shown. can be any of alerts-exposed, alerts-closed, baseline, posture or schedule. - ''' - if filename == 'alerts-exposed': + """ + if filename == "alerts-exposed": return M.tojson(M.getAlertsExposed()) - elif filename == 'alerts-closed': + elif filename == "alerts-closed": return M.tojson(M.getAlertsClosed()) - elif filename == 'baseline': + elif filename == "baseline": return M.tojson(M.getNeatline()) - elif filename == 'posture': + elif filename == "posture": return M.tojson(M.getForensic()) - elif filename == 'schedule': + elif filename == "schedule": return M.tojson(SM.getSchedule()) else: abort(404, "undefined json") @@ -174,16 +194,16 @@ def returnjson(filename): # JSON Views -@route('/plain/') +@route("/plain/") def returnplain(filename): - ''' + """ Presents all plaintext views: http:///plain/*. These are used by other scripts, like markusk's openvas-config-script @param filename the plaintext view to be shown. - ''' - response.content_type = 'text/plain' - if filename == 'scanned-ranges': + """ + response.content_type = "text/plain" + if filename == "scanned-ranges": return SM.getScannedRanges() else: abort(404, "undefined url") @@ -192,25 +212,27 @@ def returnplain(filename): # controller # rescan -@route('/controller/rescanjob/') +@route( + "/controller/rescanjob/" +) def serve_rescanController(jobid): - ''' + """ Activates controller: http:///controller/rescanjob/. Instructs the controller to perform a forensic rescan of the job with id now. @param jobid the job ID to be scanned. - ''' + """ rv = Controller.rescanJob(jobid) closeDB() return rv -@route('/controller/rescanmachine/') -@route('/controller/rescanmachine//') +@route("/controller/rescanmachine/") +@route("/controller/rescanmachine//") def serve_rescanMachine(host, port=None): - ''' + """ Activates controller: http:///controller/rescanmachine/[/]. Instructs the controller to perform a forensic rescan of the machine with id now. @@ -218,66 +240,70 @@ def serve_rescanMachine(host, port=None): @param host the host ID to be scanned. @param port the port to be rescanned on the specified machine. - ''' + """ assert host.isdigit() rv = Controller.rescanMachine(int(host), int(port) if port else None) closeDB() return rv -@route('/controller/rescanservice/') +@route("/controller/rescanservice/") def serve_rescanService(serviceid): - ''' + """ Activates controller: http:///controller/rescanservice/. Instructs the controller to perform a forensic rescan of the service with id now. @param serviceid The Service ID to be scanned. - ''' + """ assert serviceid.isdigit() rv = Controller.rescanService(int(serviceid)) closeDB() return rv -@route('/controller/deletemachine/') + +@route("/controller/deletemachine/") def serve_deleteMachine(machineid): - ''' + """ Activates controller: http:///controller/deletemachine/. Deletes the machine and all its associated services from inventory. @param machineid The Machine ID to be deleted. - ''' + """ assert machineid.isdigit() machine = Machine(int(machineid)) - + # Delete all services first for service in machine.getServices(): service.delete() - + # Then delete the machine machine.delete() - + closeDB() return "ok" -@route('/controller/deletejob/') + +@route( + "/controller/deletejob/" +) def serve_deleteJob(jobid): - ''' + """ Activates controller: http:///controller/deletejob/. Instructs the controller to delete the job with id . @param jobid The Job ID to be scanned. - ''' + """ rv = SM.deleteJob(jobid) closeDB() return rv # flipcrit -@route('/controller/flipcrit//') +@route("/controller/flipcrit//") def serve_flipCrit(page, serviceid): - ''' + """ Activates controller: http:///controller/flipcrit//. Instructs the controller to flip the criticality of service on @@ -288,54 +314,59 @@ def serve_flipCrit(page, serviceid): @param page Either "exposed" or "closed". Denominates the view on which the criticality of the service should be flipped. @param serviceid The Service whose criticality should be flipped. - ''' + """ exposed = True if page == "exposed" else False Controller.flipCrit(serviceid, exposed) closeDB() + # approve -@post('/controller/approve') +@post("/controller/approve") def serve_approve(): - ''' + """ Activates controller: http:///controller/approve. The arguments are to be passed by using the HTTP POST method. Using this method, a service can be approved to the baseline. @param pk The Service ID to be approved. @param value a business justification for the service to be approved. - ''' - serviceid = request.forms.get('pk') - justification = request.forms.get('value') + """ + serviceid = request.forms.get("pk") + justification = request.forms.get("value") Controller.approve(int(serviceid), justification, M.username) closeDB() + # remove -@post('/controller/remove') +@post("/controller/remove") def serve_remove(): - ''' + """ Activates controller: http:///controller/remove. The arguments are to be passed by using the HTTP POST method. Using this method, a service can be removed from the baseline. @param pk The Service ID to be removed. @param value a business justification for the service to be removed. - ''' - serviceid = request.forms.get('pk') - justification = request.forms.get('value') + """ + serviceid = request.forms.get("pk") + justification = request.forms.get("value") Controller.remove(int(serviceid), justification, M.username) closeDB() + # edit job -@post('/controller/editjob/') +@post( + "/controller/editjob/" +) def serve_changejob(jobid): - ''' + """ Activates controller: http:///controller/editjob/. This method tells the controller to set or change the parameters of the specified job. @@ -349,23 +380,27 @@ def serve_changejob(jobid): @param portrange a single port or port range in the format - to be scanned. @param extraparams extra command line parameters for nmap. - ''' - rv = SM.changeJob(jobid=jobid, - cronval=request.forms.get('cronval'), - iprange=request.forms.get('iprange'), - portrange=request.forms.get('portrange'), - extraparams=request.forms.get('extraparams') - ) + """ + rv = SM.changeJob( + jobid=jobid, + cronval=request.forms.get("cronval"), + iprange=request.forms.get("iprange"), + portrange=request.forms.get("portrange"), + extraparams=request.forms.get("extraparams"), + ) closeDB() return rv + # edit job -@post('/controller/addjob/') +@post( + "/controller/addjob/" +) def serve_addjob(jobid): - ''' + """ Activates controller: http:///controller/addjob/. This method tells the controller to set the parameters of the specified job and add it to crontab. @@ -379,28 +414,33 @@ def serve_addjob(jobid): @param portrange a single port or port range in the format - to be scanned. @param extraparams extra command line parameters for nmap. - ''' - rv = SM.addJob(jobid=jobid, - cronval=request.forms.get('cronval'), - iprange=request.forms.get('iprange'), - portrange=request.forms.get('portrange'), - extraparams=request.forms.get('extraparams') - ) + """ + rv = SM.addJob( + jobid=jobid, + cronval=request.forms.get("cronval"), + iprange=request.forms.get("iprange"), + portrange=request.forms.get("portrange"), + extraparams=request.forms.get("extraparams"), + ) closeDB() return rv # static files -@route('/' + sr + '') +@route("/" + sr + "") def static(filename): - ''' + """ returns static files from the path defined by variable SR. @param filename path to the static file relative to the SR directory. - ''' + """ return static_file(filename, root=sr) # run the service! -run(host=localconf['server']['listen'], port=localconf['server']['port'], - debug=localconf['server']['debug'], server='paste') +run( + host=localconf["server"]["listen"], + port=localconf["server"]["port"], + debug=localconf["server"]["debug"], + server="paste", +) diff --git a/aspromMetrics.py b/aspromMetrics.py index 3388054..6c1bbd4 100644 --- a/aspromMetrics.py +++ b/aspromMetrics.py @@ -1,32 +1,50 @@ -''' +""" Created on Sep 05, 2024 @author stefankn -@namespace asprom.aspromNagiosCheck +@namespace asprom.aspromMetrics small and nice metrics server for asprom -''' -from inc.asprom import initDB, closeDB, AspromModel, Cfg +""" + from time import sleep -from prometheus_client import start_http_server, Gauge -from pprint import pprint -localconf = Cfg() +from prometheus_client import Gauge, start_http_server -alertsExposed = Gauge('alerts_exposed', 'These Ports are unintentionally open and therefore to be checked with the highest priority.') -alertsClosed = Gauge('alerts_closed', 'These Ports are unintentionally open and therefore to be checked with the highest priority.') +from inc.asprom import AspromModel, Cfg, initDB +from inc.logging import configure_logging, get_logger -initDB(localconf) -M = AspromModel() +configure_logging() +logger = get_logger(__name__) + +alertsExposed = Gauge( + "alerts_exposed", + "These Ports are unintentionally open and therefore to be checked with the highest priority.", +) +alertsClosed = Gauge( + "alerts_closed", + "These Ports are unintentionally open and therefore to be checked with the highest priority.", +) + +M = None -def refreshMetrics(): - alertsExposed.set(len(M.getAlertsExposed())) - alertsClosed.set(len(M.getAlertsClosed())) +def _ensure_model(): + global M + if M is None: + initDB(Cfg()) + M = AspromModel() + return M + + +def refreshMetrics(): + model = _ensure_model() + alertsExposed.set(len(model.getAlertsExposed())) + alertsClosed.set(len(model.getAlertsClosed())) -if __name__ == '__main__': - pprint("starting asprom metrics server") - # Start up the server to expose the metrics. +if __name__ == "__main__": + logger.info("starting asprom metrics server") + _ensure_model() start_http_server(5000) while True: diff --git a/aspromNagiosCheck.py b/aspromNagiosCheck.py index 6d44594..111a604 100644 --- a/aspromNagiosCheck.py +++ b/aspromNagiosCheck.py @@ -1,58 +1,54 @@ -''' +""" Created on Oct 23, 2014 @author stefankn @namespace asprom.aspromNagiosCheck This file is invoked from the CLI and can be directly used as a nagios plugin. -if any of the services on the alerts-exposed or alerts-closed views -are marked as critical, -this script terminates with a return value of 2. -if none are marked as critical, but at least one is marked as warning, -this script terminates with a return value of 1. -Else, it terminates with a value of 0 signalling everything is alright. -''' -from inc.asprom import initDB, closeDB, AspromModel, Cfg, genMessages +""" + import sys +from inc.asprom import AspromModel, Cfg, closeDB, genMessages, initDB +from inc.logging import configure_logging, get_logger + +configure_logging() +logger = get_logger(__name__) + + def main(): exitstate = 0 msg = "" localconf = Cfg() initDB(localconf) - M = AspromModel() - - #exposed services - messageCritExposed, messageWarnExposed = genMessages(M.getAlertsExposed()) + model = AspromModel() - #closed services - messageCritClosed, messageWarnClosed = genMessages(M.getAlertsClosed()) + message_crit_exposed, message_warn_exposed = genMessages(model.getAlertsExposed()) + message_crit_closed, message_warn_closed = genMessages(model.getAlertsClosed()) closeDB() - # Start up the server to expose the metrics. - start_http_server(5000) - if len(messageCritExposed): - msg += 'CRITICAL-EXPOSED: ' + " | ".join(messageCritExposed) + "\n" + if len(message_crit_exposed): + msg += "CRITICAL-EXPOSED: " + " | ".join(message_crit_exposed) + "\n" exitstate = 2 - if len(messageCritClosed): - msg += 'CRITICAL-CLOSED: ' + " | ".join(messageCritClosed) + "\n" + if len(message_crit_closed): + msg += "CRITICAL-CLOSED: " + " | ".join(message_crit_closed) + "\n" exitstate = 2 - if len(messageWarnExposed): - msg += 'WARNING-EXPOSED: ' + " | ".join(messageWarnExposed) + "\n" + if len(message_warn_exposed): + msg += "WARNING-EXPOSED: " + " | ".join(message_warn_exposed) + "\n" exitstate = exitstate or 1 - if len(messageWarnClosed): - msg += 'WARNING-CLOSED: ' + " | ".join(messageWarnClosed) + "\n" + if len(message_warn_closed): + msg += "WARNING-CLOSED: " + " | ".join(message_warn_closed) + "\n" exitstate = exitstate or 1 if not exitstate: - msg = 'all Profiles nominal.' + msg = "all Profiles nominal." - msg += 'Profiling URL: ' + localconf['misc']['url'] + msg += "Profiling URL: " + localconf["misc"]["url"] - print(msg) + logger.info("nagios_check_complete", exitstate=exitstate, message=msg) sys.exit(exitstate) -if __name__ == '__main__': +if __name__ == "__main__": main() diff --git a/aspromScan.py b/aspromScan.py index a5db637..0265455 100644 --- a/aspromScan.py +++ b/aspromScan.py @@ -1,67 +1,53 @@ -''' +""" Created on Oct 23, 2014 @author stefankn @namespace asprom.aspromScan this file is invoked on the CLI as a wrapper script to nmap. -when invoked from the command line, the scan() method is called. -''' -import argparse -import re -from inc.asprom import scan, initDB, closeDB, Cfg - - -def main(): - ''' - parse arguments from command line. - - -> usage: - - aspromScan.py [-h] [-o EXTRA_OPTIONS] [-s SENSOR] [-p PORT_RANGE] - [-j JOB_ID] TARGET +""" - Scans an IP Range for asprom. Needs nmap installed on the sensor host. - - - positional arguments: - TARGET the hostname/ip/ip range to be scanned - - optional arguments: - - -h, --help show this help message and exit +import argparse - -o EXTRA_OPTIONS, --extra-options EXTRA_OPTIONS - extra options to be passed to nmap +from inc.asprom import Cfg, closeDB, initDB, scan +from inc.logging import configure_logging, get_logger - -s SENSOR, --sensor SENSOR - start scanning on another sensor +configure_logging() +logger = get_logger(__name__) - -p PORT_RANGE, --port-range PORT_RANGE - set custom port range to be scanned - -j JOB_ID, --job-id JOB_ID - set arbitrary job id (used by aspromGUI and cron) - ''' - parser = argparse.ArgumentParser(description='''Scans an IP Range for - asprom. Needs nmap installed on the sensor host.''') - parser.add_argument('target', metavar="TARGET", - help='the hostname/ip/ip range to be scanned') - parser.add_argument('-o', '--extra-options', default='', - help='extra options to be passed to nmap') - parser.add_argument('-s', '--sensor', default='localhost', - help='start scanning on another sensor') - parser.add_argument('-p', '--port-range', default=None, - help='set custom port range to be scanned') - parser.add_argument('-j', '--job-id', default=None, - help='set arbitrary job id (used by aspromGUI and cron)') +def main(): + parser = argparse.ArgumentParser( + description="Scans an IP Range for asprom. Needs nmap installed on the sensor host." + ) + parser.add_argument("target", metavar="TARGET", help="the hostname/ip/ip range to be scanned") + parser.add_argument( + "-o", "--extra-options", default="", help="extra options to be passed to nmap" + ) + parser.add_argument( + "-s", "--sensor", default="localhost", help="start scanning on another sensor" + ) + parser.add_argument( + "-p", "--port-range", default=None, help="set custom port range to be scanned" + ) + parser.add_argument( + "-j", "--job-id", default=None, help="set arbitrary job id (used by aspromGUI and cron)" + ) args = parser.parse_args() + logger.info( + "scan_start", + target=args.target, + port_range=args.port_range, + job_id=args.job_id, + sensor=args.sensor, + ) localconf = Cfg() initDB(localconf) - scan(args.target, args.port_range, args.extra_options, args.job_id) + state = scan(args.target, args.port_range, args.extra_options, args.job_id) closeDB() + logger.info("scan_complete", state=state) + -if __name__ == '__main__': +if __name__ == "__main__": main() diff --git a/docker/start.sh b/docker/start.sh index c5a5316..6870825 100755 --- a/docker/start.sh +++ b/docker/start.sh @@ -1,3 +1,6 @@ #!/bin/bash +if [ "${ASPROM_RUN_MIGRATIONS:-}" = "1" ]; then + alembic upgrade head +fi cron python3 aspromGUI.py diff --git a/etc/asprom.cfg.prod-local.example b/etc/asprom.cfg.prod-local.example new file mode 100644 index 0000000..890d52e --- /dev/null +++ b/etc/asprom.cfg.prod-local.example @@ -0,0 +1,17 @@ +# Example config for ./scripts/dev-prod.sh (generated at runtime). +# Do not commit real credentials. The dev script writes a temp file and sets ASPROM_CFG. +db: { + 'host': '127.0.0.1' + 'port': 3307 + 'user': 'asprom' + 'passwd': 'REPLACE_ME' + 'db': 'asprom' +} +server: { + 'listen': '127.0.0.1' + 'port': 8080 + 'debug': True +} +misc: { + 'url': 'http://127.0.0.1:8080' +} diff --git a/inc/asprom.py b/inc/asprom.py index cd632dd..bdc5490 100644 --- a/inc/asprom.py +++ b/inc/asprom.py @@ -1,127 +1,143 @@ -''' +""" Created on Oct 22, 2014 @author stefankn @namespace asprom.inc.asprom Library for asprom Scripts. -''' +""" +from __future__ import annotations + +import copy +import os import re import socket import traceback -import copy from datetime import datetime -from config import Config -from os import path -from bottle import response, request from json import dumps +from os import path +from typing import Any + +import MySQLdb as mdb +from anyascii import anyascii +from bottle import response +from config import Config from crontab import CronTab -from netaddr import IPAddress, IPNetwork, AddrFormatError +from netaddr import AddrFormatError, IPAddress, IPNetwork from nmap import nmap -from anyascii import anyascii -import MySQLdb as mdb +from inc.db import close_db, get_cfg, get_db, init_db +from inc.logging import get_logger -class NoJoibIDException(Exception): +closeDB = close_db +initDB = init_db + +logger = get_logger(__name__) + + +class NoJobIdException(Exception): """ Exception raised for crontab entries not concerning asprom. """ def __init__(self, job): self.job = job - super(NoJoibIDException, self).__init__() + super().__init__() + + +NoJoibIDException = NoJobIdException class Cfg(Config): - ''' + """ Configuration in dict form from the config file etc/asprom.cfg. - ''' + """ + maindir = None def __init__(self): - ''' + """ expanded constructor, calls the super constructor of Config with the path to asprom.cfg - ''' - maindir = path.normpath(path.join(path.dirname(path.realpath(__file__) - ), path.pardir)) - # read config file - super(Cfg, self).__init__(maindir + '/etc/asprom.cfg') + """ + maindir = path.normpath(path.join(path.dirname(path.realpath(__file__)), path.pardir)) + config_path = os.environ.get("ASPROM_CFG", path.join(maindir, "etc", "asprom.cfg")) + super().__init__(config_path) self.maindir = maindir -class AspromModel(object): - ''' +class AspromModel: + """ This Model abstracts calls to the database. It returns rows of data for the views in the GUI. Using the toJSON static method, they can be easily converted to the json format used by bootstrap-table AJAX calls. - ''' + """ ## logged in username, e.g. by apache auth_basic username = None - def __init__(self, username=None, * args, **kwargs): - ''' + def __init__(self, username=None, *args, **kwargs): + """ standard constructor - ''' + """ self.username = username if username else "" - super(AspromModel, self).__init__(*args, **kwargs) + super().__init__(*args, **kwargs) def getAlertsExposed(self): - ''' + """ returns row data for the alerts-exposed view. @return row data for the alerts-exposed view. - ''' - #db connection - cur = request.db.cursor(mdb.cursors.DictCursor) + """ + # db connection + cur = get_db().cursor(mdb.cursors.DictCursor) q = """SELECT id, hostname, ip, port, product service, version, extrainfo, ffdate date, crit FROM exposed""" cur.execute(q) rows = cur.fetchall() - request.db.commit() + get_db().commit() for row in rows: - row['date'] = datetime.strftime(row['date'], "%Y-%m-%d %H:%M") - row['crit'] = False if 'crit' in row and row['crit'] else True + row["date"] = datetime.strftime(row["date"], "%Y-%m-%d %H:%M") + row["crit"] = False if "crit" in row and row["crit"] else True - if len(row['version']): - row['service'] = "%s (%s)" % (row['service'], row['version']) + if len(row["version"]): + row["service"] = "%s (%s)" % (row["service"], row["version"]) return rows def getAlertsClosed(self): - ''' + """ returns row data for the alerts-closed view. @return returns row data for the alerts-closed view. - ''' - #db connection - cur = request.db.cursor(mdb.cursors.DictCursor) + """ + # db connection + cur = get_db().cursor(mdb.cursors.DictCursor) q = """SELECT id, hostname, ip, port, product service, version, extrainfo, approvaldate date, justification, crit FROM closed""" cur.execute(q) rows = cur.fetchall() - request.db.commit() + get_db().commit() for row in rows: - row['date'] = datetime.strftime(row['date'], "%Y-%m-%d %H:%M") - row['crit'] = False if 'crit' in row and row['crit'] else True + row["date"] = datetime.strftime(row["date"], "%Y-%m-%d %H:%M") + row["crit"] = False if "crit" in row and row["crit"] else True - if len(row['version']): - row['service'] = "%s (%s)" % (row['service'], row['version']) + if len(row["version"]): + row["service"] = "%s (%s)" % (row["service"], row["version"]) return rows def getNeatline(self): - ''' + """ returns row data for the baseline view. @return returns row data for the baseline view. - ''' - cur = request.db.cursor(mdb.cursors.DictCursor) + """ + cur = get_db().cursor(mdb.cursors.DictCursor) q = """select m.ip, m.hostname, s.id, s.port, s.machineId, s.product service, s.version, s.extrainfo, n.justification, n.date from services s @@ -134,19 +150,19 @@ def getNeatline(self): rows = cur.fetchall() for row in rows: - row['date'] = datetime.strftime(row['date'], "%Y-%m-%d %H:%M") + row["date"] = datetime.strftime(row["date"], "%Y-%m-%d %H:%M") - if len(row['version']): - row['service'] = "%s (%s)" % (row['service'], row['version']) + if len(row["version"]): + row["service"] = "%s (%s)" % (row["service"], row["version"]) return rows def getForensic(self): - ''' + """ returns row data for the forensic view. @return returns row data for the forensic view. - ''' - cur = request.db.cursor(mdb.cursors.DictCursor) + """ + cur = get_db().cursor(mdb.cursors.DictCursor) q = """select m.ip, m.hostname, s.id, s.port, s.machineId, s.product service, s.version, s.extrainfo, s.ffdate date from services s @@ -159,67 +175,77 @@ def getForensic(self): rows = cur.fetchall() for row in rows: - row['date'] = datetime.strftime(row['date'], "%Y-%m-%d %H:%M") + row["date"] = datetime.strftime(row["date"], "%Y-%m-%d %H:%M") - if len(row['version']): - row['service'] = "%s (%s)" % (row['service'], row['version']) + if len(row["version"]): + row["service"] = "%s (%s)" % (row["service"], row["version"]) return rows - def getLastLog(self, count): - ''' + def getLastLog(self, count: int) -> str: + """ returns last log entries from the changelog in html format. @param count: number of lines to return. @return last log entries from the changelog in html format. - ''' - cur = request.db.cursor(mdb.cursors.DictCursor) + """ + cur = get_db().cursor(mdb.cursors.DictCursor) q = """select c.date, c.neat, s.port, s.product, m.ip, m.hostname, c.justification, c.username from changelog c inner join services s on c.serviceId = s.id inner join machines m on s.machineId = m.id order by c.id desc - limit %d - """ % count - cur.execute(q) + limit %s""" + cur.execute(q, (count,)) rows = cur.fetchall() html = "" for row in rows: - date = datetime.strftime(row['date'], "%Y-%m-%d %H:%M") - port = "%s" % row['port'] - ip = "%s" % row['ip'] - just = "%s" % row['justification'] - - html += (date + ":" + (row['username'] if row['username'] else "")\ - + " " + ("added" if row['neat'] else "removed")) + \ - ((" service %s[%s]" % (row['product'], port)) if row["product"\ - ] else (" port %s" % port)) + " on host " + ("%s[%s]" % (row[ \ - 'hostname'], ip) if row["hostname"] else ip) + \ - ' with justification "%s".
' % just + date = datetime.strftime(row["date"], "%Y-%m-%d %H:%M") + port = "%s" % row["port"] + ip = "%s" % row["ip"] + just = "%s" % row["justification"] + + html += ( + ( + date + + ":" + + (row["username"] if row["username"] else "") + + " " + + ("added" if row["neat"] else "removed") + ) + + ( + (" service %s[%s]" % (row["product"], port)) + if row["product"] + else (" port %s" % port) + ) + + " on host " + + ("%s[%s]" % (row["hostname"], ip) if row["hostname"] else ip) + + ' with justification "%s".
' % just + ) return html @staticmethod def tojson(someDict): - ''' + """ converts row data to json format. @param someDict: row data in dictionary format. @return JSON String. - ''' - response.content_type = 'application/json' + """ + response.content_type = "application/json" return dumps(someDict) class AspromScheduleModel(CronTab): - ''' + """ This model abstracts access to the schedule, which consists of the users crontab and log data in the mysql DB. Both components are joined together using an UUID, the jobID. requires CronTab.py, as this class inherits from that. - ''' + """ ## schedule log from database scheduleLog = None @@ -228,87 +254,89 @@ class AspromScheduleModel(CronTab): schedule = None ## dictionary of jobs indexed by id - jobsByID = dict() + jobsByID: dict[str, Any] = {} def __init__(self, *args, **kwargs): - ''' + """ standard Constructor - ''' - super(AspromScheduleModel, self).__init__(*args, **kwargs) + """ + super().__init__(*args, **kwargs) self.read() def read(self, filename=None): - ''' + """ override read method in CronTab.py. Additionally fetches log information from the database and fills the properties schedule and scheduleLog. - ''' - super(AspromScheduleModel, self).read(filename=filename) + """ + super().read(filename=filename) self.scheduleLog = self.__fetchScheduleLog() self.schedule = self.__fetchSchedule() @staticmethod - def promoteToIndex(dici, valueKey): - ''' - index the dic by valueKey. - promotes the element on position from each sublist to an - index in a dictionary. - - @param dic a list of lists or a list of dictionaries, e.g. a - database result set. - @param valueKey position or name of the value to be promoted to - an index. - - @return promoted dictionary. - -> example: - - >>> d = [[1,2,3,4,5], [6,7], [8,9], [9,10]] - >>> e=AspromScheduleModel.promoteToIndex(d,1) - >>> print e - {9: [8], 2: [1, 3, 4, 5], 10: [9], 7: [6]} - ''' + def promoteToIndex( + dici: list[dict[str, Any]] | list[list[Any]], valueKey: str | int + ) -> dict[Any, Any]: + """ + index the dic by valueKey. + promotes the element on position from each sublist to an + index in a dictionary. + + @param dic a list of lists or a list of dictionaries, e.g. a + database result set. + @param valueKey position or name of the value to be promoted to + an index. + + @return promoted dictionary. + + > example: + + >>> d = [[1,2,3,4,5], [6,7], [8,9], [9,10]] + >>> e=AspromScheduleModel.promoteToIndex(d,1) + >>> print e + {9: [8], 2: [1, 3, 4, 5], 10: [9], 7: [6]} + """ dic = copy.deepcopy(dici) rv = {} for row in dic: - rv[row.pop(valueKey)] = row + key = row.pop(valueKey) # type: ignore[arg-type] + rv[key] = row return rv def __fetchScheduleLog(self): - ''' + """ returns last log entry of past runs for every defined job from the database. @return dictionary of jobs with log information. - ''' - dbc = request.db.cursor(mdb.cursors.DictCursor) + """ + dbc = get_db().cursor(mdb.cursors.DictCursor) # get last log entry for each job - q = '''select s.jobid, state, startdate, enddate, output from scanlog s + q = """select s.jobid, state, startdate, enddate, output from scanlog s inner join ( select jobid, max(startdate) as maxstartdate from scanlog group by jobid ) gs on s.jobid = gs.jobid and gs.maxstartdate = s.startdate - order by id asc''' + order by id asc""" dbc.execute(q) rows = dbc.fetchall() - #index the result by jobid - return self.promoteToIndex(rows, 'jobid') + # index the result by jobid + return self.promoteToIndex(rows, "jobid") def getJobByID(self, jobid): - ''' + """ returns the job with id . @param jobid: the job's id. @return a job object. - ''' + """ return self.jobsByID[jobid] - def changeJob(self, jobid, cronval, iprange, portrange, extraparams, - job=None): - ''' + def changeJob(self, jobid, cronval, iprange, portrange, extraparams, job=None): + """ changes the parameters of the Job with UUID in the crontab. @param jobid: the job to be edited. @param cronval: the cron schedule string. @@ -316,44 +344,48 @@ def changeJob(self, jobid, cronval, iprange, portrange, extraparams, @param portrange: a single port or port range in the format - to be scanned. @param extraparams: extra command line parameters for nmap. - ''' + """ - print("Controller.changeJob Input: ") - print("jobid: " + jobid) - print("cronval: " + cronval) - print("iprange: " + iprange) - print("portrange: " + portrange) - print("extraparams: " + extraparams) + logger.debug( + "Controller.changeJob input", + jobid=jobid, + cronval=cronval, + iprange=iprange, + portrange=portrange, + extraparams=extraparams, + ) # parameter assertions - assert re.match( - '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', - jobid) + assert re.match("^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", jobid) assert len(cronval) > 5 - assert re.match(r'^([^-][^\s]+)$', iprange) - assert re.match(r'^[\d-]*$', portrange) + assert re.match(r"^([^-][^\s]+)$", iprange) + assert re.match(r"^[\d-]*$", portrange) assert not re.match('["]', extraparams) if not job: job = self.getJobByID(jobid) job.setall(cronval) - job.set_command('python %s/aspromScan.py -j %s %s%s%s' % - (request.cfg.maindir, jobid, - '-o="%s" ' % extraparams if extraparams else "", - "-p %s " % portrange if portrange else "", - iprange)) - job.set_comment('asprom %s' % datetime.now().strftime("%Y-%m-%d %H:%M" - )) + job.set_command( + "python %s/aspromScan.py -j %s %s%s%s" + % ( + get_cfg().maindir, + jobid, + '-o="%s" ' % extraparams if extraparams else "", + "-p %s " % portrange if portrange else "", + iprange, + ) + ) + job.set_comment("asprom %s" % datetime.now().strftime("%Y-%m-%d %H:%M")) job.enable() self.render() - print("job enabled: " + str(job.is_enabled())) + logger.info("job enabled", enabled=job.is_enabled()) self.write() self.read() def addJob(self, jobid, cronval, iprange, portrange, extraparams): - ''' + """ adds a Job to the crontab. @param jobid: the job id to be added. @param cronval: the cron schedule string. @@ -361,41 +393,39 @@ def addJob(self, jobid, cronval, iprange, portrange, extraparams): @param portrange: a single port or port range in the format - to be scanned. @param extraparams: extra command line parameters for nmap. - ''' + """ job = self.new("/bin/true") try: - self.changeJob(jobid, cronval, iprange, portrange, extraparams, - job) + self.changeJob(jobid, cronval, iprange, portrange, extraparams, job) except AssertionError: job.clear() raise def deleteJob(self, jobid): - ''' + """ deactivates the job with id and refreshes the model. @param jobid: the job's id. - ''' + """ job = self.getJobByID(jobid) job.enable(False) - job.set_comment('asprom %s' % - datetime.now().strftime("%Y-%m-%d %H:%M")) + job.set_comment("asprom %s" % datetime.now().strftime("%Y-%m-%d %H:%M")) self.write() self.read() def getScheduleEntryByID(self, jobid): - ''' + """ returns the job specifics for the job with id . @param jobid: the job's id. @return a dictionary with job parameters. - ''' + """ return self.__getScheduleI()[jobid] def __fetchJob(self, job): - ''' + """ fetches the job specifics from crontab and database. @param job: the job to be parsed. @@ -403,28 +433,28 @@ def __fetchJob(self, job): lastrun,nextrun,laststate,params,ports. @raise NoJobIDException: will be raised if a job is encountered which has no job id, e.g. a non-asprom crontab entry. - ''' + """ # get id - m = re.search(r'-j[= ]([^\s]+) ', job.command) - if (m and job.is_enabled()): + m = re.search(r"-j[= ]([^\s]+) ", job.command) + if m and job.is_enabled(): uuidx = m.group(1) # get sensor - m = re.search(r'-s[= ]([^\s]+) ', job.command) + m = re.search(r"-s[= ]([^\s]+) ", job.command) if m: sensor = m.group(1) else: sensor = "localhost" # get ip range - m = re.search(r'[^-][^\s]+ ([^-][^\s]+)$', job.command) + m = re.search(r"[^-][^\s]+ ([^-][^\s]+)$", job.command) if m: iprange = m.group(1) else: iprange = "invalid" # port range - m = re.search(r'-p[= ]([^\s]+) ', job.command) + m = re.search(r"-p[= ]([^\s]+) ", job.command) if m: ports = m.group(1) else: @@ -443,16 +473,19 @@ def __fetchJob(self, job): lastLog = self.scheduleLog[uuidx] # last state - if 'enddate' in lastLog and lastLog['enddate'] is not None: - laststate = lastLog['state'] + '(' + (lastLog['enddate'] - - lastLog['startdate']).__str__() + ('h)') + if "enddate" in lastLog and lastLog["enddate"] is not None: + laststate = ( + lastLog["state"] + + "(" + + (lastLog["enddate"] - lastLog["startdate"]).__str__() + + ("h)") + ) else: - laststate = lastLog['state'] + laststate = lastLog["state"] # start date - if 'startdate' in lastLog or lastLog['startdate'] is not None: - startdate = datetime.strftime(lastLog['startdate'], - "%Y-%m-%d %H:%M") + if "startdate" in lastLog or lastLog["startdate"] is not None: + startdate = datetime.strftime(lastLog["startdate"], "%Y-%m-%d %H:%M") else: startdate = "-" @@ -460,87 +493,96 @@ def __fetchJob(self, job): laststate = "-" startdate = "-" - return ({"id": uuidx, "when": job.slices.render(), "iprange": - iprange, "sensor": sensor, "lastrun": startdate, "nextrun": - datetime.strftime(job.schedule().get_next(), "%Y-%m-%d %H:%M"), - "laststate": laststate, "params": params, "ports": ports}) - #uuid().__str__() + return { + "id": uuidx, + "when": job.slices.render(), + "iprange": iprange, + "sensor": sensor, + "lastrun": startdate, + "nextrun": datetime.strftime(job.schedule().get_next(), "%Y-%m-%d %H:%M"), + "laststate": laststate, + "params": params, + "ports": ports, + } + # uuid().__str__() else: # this is no crontab entry for asprom. - raise NoJoibIDException( - 'not a crontab entry for asprom, no jobid found: %s' % (job)) + raise NoJobIdException("not a crontab entry for asprom, no jobid found: %s" % (job)) def getSchedule(self): - ''' + """ returns the schedule (crontab) in flat (unindexed) form, e.g. for GUI table data. - ''' + """ return self.schedule + def getScannedRanges(self) -> str: + return "\n".join(job["iprange"] for job in self.getSchedule()) + def __getScheduleI(self): - ''' + """ returns the schedule (crontab) indexed by jobid. - ''' - return self.promoteToIndex(self.schedule, 'id') + """ + return self.promoteToIndex(self.schedule, "id") def __fetchSchedule(self): - ''' + """ returns row data for the schedule view. - ''' + """ rv = [] - #reread crontab from disk - #self.read() + # reread crontab from disk + # self.read() # extract nmap arguments from crontab definition by regular expressions for job in self: try: - #get the job specifics + # get the job specifics jobSpec = self.__fetchJob(job) rv.append(jobSpec) # also fill up the jobsByID attribute - jobid = jobSpec['id'] + jobid = jobSpec["id"] self.jobsByID[jobid] = job - except NoJoibIDException: + except NoJobIdException: pass return rv -class Controller(object): - ''' +class Controller: + """ The Controller defines all actions that are possible from within the GUI. - ''' + """ @staticmethod def rescanJob(jobid): - ''' + """ run the scheduled job with id right now. @param jobid: the UUID of the job to be run - ''' + """ # get job schedule from cron SM = AspromScheduleModel(user=True) - jobs = SM.promoteToIndex(SM.getSchedule(), 'id') + jobs = SM.promoteToIndex(SM.getSchedule(), "id") if jobs[jobid]: j = jobs[jobid] - print("rescanning job: " + str(j)) - ps = scan(j['iprange'], j['ports'], j['params'], jobid) + logger.info("rescanning job", job=j) + ps = scan(j["iprange"], j["ports"], j["params"], jobid) return ps @staticmethod def rescanMachine(machineid, port=None): - ''' + """ rescan the machine. this method finds the task in the schedule to which the machine belongs and takes its additional arguments from there. @param machineid: ID of the machine to be rescanned. @param port: port of the machine to be rescanned. - ''' + """ # get job schedule from cron sm = AspromScheduleModel(user=True) @@ -554,36 +596,43 @@ def rescanMachine(machineid, port=None): # first, check if iprange is an actual ip range and ip is in that # range. try: - if machine.ip in IPNetwork(job['iprange']): - jobid = job['id'] - print("IP %s in range %s - jobid %s" % (machine.ip, job[ - 'iprange'], job['id'])) + if machine.ip in IPNetwork(job["iprange"]): + jobid = job["id"] + logger.debug( + "IP in range", + ip=str(machine.ip), + iprange=job["iprange"], + jobid=job["id"], + ) break except AddrFormatError as e: - print("%s - trying by name resolution" % e) + logger.debug("addr format error, trying name resolution", error=str(e)) # if not, check if iprange resolves to the given machine try: - if str(machine.ip) == socket.gethostbyname(job['iprange']): - jobid = job['id'] - print("IP %s == %s - jobid %s" % (machine.ip, job[ - 'iprange'], job['id'])) + if str(machine.ip) == socket.gethostbyname(job["iprange"]): + jobid = job["id"] + logger.debug( + "IP matched by name resolution", + ip=str(machine.ip), + iprange=job["iprange"], + jobid=job["id"], + ) break - except Exception as e: - print("shit happened : %s" % e) + except (OSError, socket.gaierror) as e: + logger.warning("name resolution failed", error=str(e)) if jobid: - ps = scan(str(machine.ip), str(port) if port else job['ports'], - job['params'], jobid) + ps = scan(str(machine.ip), str(port) if port else job["ports"], job["params"], jobid) return ps @staticmethod def rescanService(serviceid): - ''' + """ rescans the service. @param serviceid: ID of the service to be rescanned. - ''' + """ # get machine details from database serv = Service(serviceid) machine = serv.getMachine() @@ -592,7 +641,7 @@ def rescanService(serviceid): @staticmethod def flipCrit(serviceid, exposed=True): - ''' + """ flips the criticality of the service. Flipping sets the service criticality to WARNING if it was CRITICAL before and the other way round. @@ -601,41 +650,41 @@ def flipCrit(serviceid, exposed=True): @param page Denominates the view on which the criticality of the service should be flipped. If true, "alerts-exposed" is flipped. Else, "alerts-closed". - ''' + """ s = Service(int(serviceid)) s.flipCrit(exposed) @staticmethod def approve(serviceid, justification, username): - ''' + """ Using this method, a service can be approved to the baseline. @param serviceid: The Service ID to be approved. @param justification: a business justification for the service to be approved. - ''' + """ s = Service(int(serviceid)) s.approve(justification, username) @staticmethod def remove(serviceid, justification, username): - ''' + """ Using this method, a service can be removed from the baseline. @param serviceid: The Service ID to be removed. @param justification: a business justification for the service to be removed. - ''' + """ s = Service(int(serviceid)) s.remove(justification, username) -class Service(object): - ''' +class Service: + """ This class represents a single Service, that is a port on a machine. - ''' + """ - #attributes + # attributes ## services database id id = None @@ -669,13 +718,13 @@ class Service(object): critClosed = False def __init__(self, serviceid): - ''' + """ Constructor. Loads all information about the service from the database. @param serviceid: Database id of the service. - ''' - cur = request.db.cursor(mdb.cursors.DictCursor) + """ + cur = get_db().cursor(mdb.cursors.DictCursor) q = """SELECT s.id id, m.id mid, port, s.ffdate, s.lsdate, s.product, s.version, s.extrainfo, c.flipExposed, c.flipClosed @@ -683,26 +732,24 @@ def __init__(self, serviceid): on m.id = s.machineId left join criticality c on s.id = c.serviceId - WHERE s.id=%d""" % serviceid - cur.execute(q) + WHERE s.id=%s""" + cur.execute(q, (serviceid,)) row = cur.fetchone() - self.id = row['id'] - self.machine = Machine(row['mid']) - self.port = row['port'] - self.product = row['product'] - self.version = row['version'] - self.extrainfo = row['extrainfo'] - self.lsdate = row['lsdate'] - self.ffdate = row['ffdate'] - self.critExposed = (True if 'flipExposed' in row and row['flipExposed'] - == 1 else False) - self.critClosed = (True if 'flipClosed' in row and row['flipClosed'] == - 1 else False) + self.id = row["id"] + self.machine = Machine(row["mid"]) + self.port = row["port"] + self.product = row["product"] + self.version = row["version"] + self.extrainfo = row["extrainfo"] + self.lsdate = row["lsdate"] + self.ffdate = row["ffdate"] + self.critExposed = True if "flipExposed" in row and row["flipExposed"] == 1 else False + self.critClosed = True if "flipClosed" in row and row["flipClosed"] == 1 else False @staticmethod - def create(mach, portno, product='', version='', extrainfo=''): - ''' + def create(mach, portno, product="", version="", extrainfo=""): + """ create new service and return self @param mach: Machine object to which the service belongs @@ -712,161 +759,149 @@ def create(mach, portno, product='', version='', extrainfo=''): @param version: additional version information gleaned by nmap @param extrainfo: additional extra information gleaned by nmap @return self - ''' + """ - cur = request.db.cursor() + cur = get_db().cursor() # if product not defined, get generic information about port from # /etc/services. if not product: try: - product = socket.getservbyport(portno, 'tcp') - except: + product = socket.getservbyport(portno, "tcp") + except OSError: pass - q = ("""INSERT INTO services (port, protocolId, machineId, product, + q = """INSERT INTO services (port, protocolId, machineId, product, extrainfo, version, lsdate, ffdate) - VALUES (%d, %d, %d, "%s", "%s", "%s", NOW(), NOW()) + VALUES (%s, %s, %s, %s, %s, %s, NOW(), NOW()) ON DUPLICATE KEY UPDATE id=LAST_INSERT_ID(id), lsdate=NOW(); """ - % (portno, 1, mach.id, product, extrainfo, version)) - print(q) - cur.execute(q) + logger.debug("inserting service", query=q) + cur.execute(q, (portno, 1, mach.id, product, extrainfo, version)) mid = cur.lastrowid - print("Service ID inserted: %s" % mid) + logger.info("service inserted", service_id=mid) ## check if last log entry is negative - q = ( - """select openp from servicelog where serviceid=%d order by id - desc limit 1""" - % mid) - cur.execute(q) + q = """select openp from servicelog where serviceid=%s order by id desc limit 1""" + cur.execute(q, (mid,)) try: rv = cur.fetchone()[0] except (TypeError, KeyError): rv = False # if it is negative, insert a positive log entry if not rv: - q = ( - """INSERT INTO servicelog (serviceId, openp, date) values - (%d, %d, NOW()) """ - % (mid, 1)) - cur.execute(q) + q = """INSERT INTO servicelog (serviceId, openp, date) values (%s, %s, NOW())""" + cur.execute(q, (mid, 1)) - request.db.commit() + get_db().commit() return Service(mid) def delete(self): - ''' + """ deletes this service. - ''' - cur = request.db.cursor() - q = """DELETE FROM criticality WHERE serviceId = %d""" % (self.id) - cur.execute(q) + """ + cur = get_db().cursor() + q = """DELETE FROM criticality WHERE serviceId = %s""" + cur.execute(q, (self.id,)) ## check if last log entry is positive - q = ( - """SELECT openp FROM servicelog WHERE serviceid=%d order by id - desc limit 1""" - % self.id) - cur.execute(q) + q = """SELECT openp FROM servicelog WHERE serviceid=%s order by id desc limit 1""" + cur.execute(q, (self.id,)) try: rv = cur.fetchone()[0] except (TypeError, KeyError): rv = False # if it is positve, insert a negative log entry if rv: - q = ( - """INSERT INTO servicelog (serviceId, openp, date) values - (%d, %d, NOW())""" - % (self.id, 0)) - cur.execute(q) - request.db.commit() + q = """INSERT INTO servicelog (serviceId, openp, date) values (%s, %s, NOW())""" + cur.execute(q, (self.id, 0)) + get_db().commit() def getMachine(self): - ''' + """ returns the machine object associated with this service. @return a machine object. - ''' + """ return self.machine - def inRange(self, r): - ''' + def inRange(self, r: str | int) -> bool: + """ tells if a service is in a specific port range. @param r: single port number or range, e.g. "1024-65535" @return boolean. - ''' - #single port? - if (isinstance(r, int) or r.isdigit()): + """ + if self.port is None: + raise Exception("service port is not set") + # single port? + if isinstance(r, int) or r.isdigit(): return int(r) == self.port - #range as string? - m = re.search(r'^(\d+)-(\d+)$', r) + # range as string? + m = re.search(r"^(\d+)-(\d+)$", r) if m: startr = m.group(1) endr = m.group(2) - return int(startr) <= self.port <= int(endr) + return int(startr) <= int(self.port) <= int(endr) else: - raise Exception('cannot determine if %s is in range %s' % (self. - port, r)) + raise Exception("cannot determine if %s is in range %s" % (self.port, r)) def flipCrit(self, exposed=True): - ''' + """ Flip criticality of "exposed" view if exposed = true, else of the "closed" view @param exposed: a boolean. - ''' + """ if exposed: - col = 'flipExposed' + col = "flipExposed" self.critExposed = not self.critExposed val = self.critExposed else: - col = 'flipClosed' + col = "flipClosed" self.critClosed = not self.critClosed val = self.critClosed - cur = request.db.cursor() - q = """INSERT INTO criticality (serviceId, %s) - VALUES (%d, %d) - ON DUPLICATE KEY UPDATE %s=%d""" % (col, self.id, val, col, val) - print(q) - cur.execute(q) - request.db.commit() + if col not in ("flipExposed", "flipClosed"): + raise ValueError(f"invalid criticality column: {col}") + cur = get_db().cursor() + q = f"""INSERT INTO criticality (serviceId, {col}) + VALUES (%s, %s) + ON DUPLICATE KEY UPDATE {col}=%s""" + logger.debug("updating criticality", query=q) + cur.execute(q, (self.id, int(val), int(val))) + get_db().commit() def approve(self, justification, username, neat=True): - ''' + """ approve this service and add it to the baseline @param justification: a business justification. @param neat: true for approval. if false, remove from baseline. this is used by the method remove(). - ''' + """ - cur = request.db.cursor() - q = ("""INSERT INTO changelog (serviceId, neat, justification, date, - username) VALUES (%d, %d, "%s", NOW(), "%s")""" - % (self.id, 1 if neat else 0, justification, username)) - cur.execute(q) - q = ("""UPDATE criticality SET flipExposed=0, flipClosed=0 - WHERE serviceId = %d""" % self.id) - cur.execute(q) - request.db.commit() + cur = get_db().cursor() + q = """INSERT INTO changelog (serviceId, neat, justification, date, + username) VALUES (%s, %s, %s, NOW(), %s)""" + cur.execute(q, (self.id, 1 if neat else 0, justification, username)) + q = """UPDATE criticality SET flipExposed=0, flipClosed=0 WHERE serviceId = %s""" + cur.execute(q, (self.id,)) + get_db().commit() def remove(self, justification, username): - ''' + """ remove this service from the baseline. @param justification: a business justification. - ''' + """ self.approve(justification, username, False) -class Machine(object): - ''' +class Machine: + """ represents a machine, that is a singular IP adress. - ''' + """ ## machines database id. id = None @@ -884,76 +919,72 @@ class Machine(object): ffdate = None def __init__(self, machineid): - ''' + """ Constructor Loads all information about the machine from the database. @param machineid: Database id of the machine. - ''' - cur = request.db.cursor(mdb.cursors.DictCursor) - q = ("""SELECT id, ip, hostname, lsdate, ffdate FROM machines - WHERE id='%d'""" % machineid) - cur.execute(q) + """ + cur = get_db().cursor(mdb.cursors.DictCursor) + q = """SELECT id, ip, hostname, lsdate, ffdate FROM machines WHERE id=%s""" + cur.execute(q, (machineid,)) row = cur.fetchone() - self.id = int(row['id']) - self.ip = IPAddress(row['ip']) - self.hostname = row['hostname'] - self.lsdate = row['lsdate'] - self.ffdate = row['ffdate'] + self.id = int(row["id"]) + self.ip = IPAddress(row["ip"]) + self.hostname = row["hostname"] + self.lsdate = row["lsdate"] + self.ffdate = row["ffdate"] @staticmethod def create(name, ip): - ''' + """ create new Machine and return self. @param name: hostname @param ip: ip address @return self - ''' - cur = request.db.cursor(mdb.cursors.DictCursor) - q = ("""INSERT INTO machines (hostname, ip, rangeId, lsdate, ffdate) - VALUES ("%s", "%s", %d, NOW(), NOW()) + """ + cur = get_db().cursor(mdb.cursors.DictCursor) + q = """INSERT INTO machines (hostname, ip, rangeId, lsdate, ffdate) + VALUES (%s, %s, %s, NOW(), NOW()) ON DUPLICATE KEY UPDATE id=LAST_INSERT_ID(id), - hostname="%s", lsdate=NOW(); """ - % (name, ip, 1, name)) + hostname=%s, lsdate=NOW(); """ - print(q) - cur.execute(q) + logger.debug("inserting machine", query=q) + cur.execute(q, (name, ip, 1, name)) mid = cur.lastrowid - print("Machine ID inserted: %s" % mid) + logger.info("machine inserted", machine_id=mid) ## check if last log entry is negative - q = ("""select exposed from machinelog where machineId=%d - order by id desc limit 1""" % mid) - cur.execute(q) + q = """select exposed from machinelog where machineId=%s order by id desc limit 1""" + cur.execute(q, (mid,)) try: - rv = cur.fetchone()['exposed'] - except (TypeError): + rv = cur.fetchone()["exposed"] + except TypeError: rv = False # if it is negative, insert a positive log entry if not rv: - q = ("""INSERT INTO machinelog (machineId, exposed, date) - values (%d, %d, NOW()) """ % (mid, 1)) - cur.execute(q) + q = """INSERT INTO machinelog (machineId, exposed, date) values (%s, %s, NOW())""" + cur.execute(q, (mid, 1)) - request.db.commit() + get_db().commit() return Machine(mid) def getServices(self, exposedOnly=False): - ''' + """ return list of services associated with this machine. @param exposedOnly: if true, only return currently exposed services. @return list of services. - ''' - cur = request.db.cursor() - q = ("""select id from services s inner join servicelogCur l - on s.id=l.serviceId where machineId = %d %s""" - % (self.id, "AND openp=1" if exposedOnly else "")) - - cur.execute(q) + """ + cur = get_db().cursor() + q = """select id from services s inner join servicelogCur l + on s.id=l.serviceId where machineId = %s """ + if exposedOnly: + q += "AND openp=1" + cur.execute(q, (self.id,)) rows = cur.fetchall() services = [] @@ -964,68 +995,73 @@ def getServices(self, exposedOnly=False): return services def delete(self): - ''' + """ Delete this machine and clean up related records. - ''' + """ - print("deleting machine %s" % self.id) - cur = request.db.cursor() + logger.info("deleting machine", machine_id=self.id) + cur = get_db().cursor() ## check if last log entry is positive - q = ("""select exposed from machinelog where machineId=%d - order by id desc limit 1""" % self.id) - cur.execute(q) + q = """select exposed from machinelog where machineId=%s order by id desc limit 1""" + cur.execute(q, (self.id,)) try: rv = cur.fetchone()[0] except (TypeError, KeyError): rv = False - # if it is positive, insert a negative log entry - if not rv: - q = ( - """INSERT INTO machinelog (machineId, exposed, date) - values (%d, %d, NOW())""" % (self.id, 0)) - cur.execute(q) + # if it was exposed, insert a negative log entry + if rv: + q = """INSERT INTO machinelog (machineId, exposed, date) values (%s, %s, NOW())""" + cur.execute(q, (self.id, 0)) - cur.execute(q) - request.db.commit() + for svc in self.getServices(): + cur.execute("DELETE FROM criticality WHERE serviceId = %s", (svc.id,)) + cur.execute("DELETE FROM servicelog WHERE serviceId = %s", (svc.id,)) + cur.execute("DELETE FROM changelog WHERE serviceId = %s", (svc.id,)) + cur.execute("DELETE FROM services WHERE id = %s", (svc.id,)) + + cur.execute("DELETE FROM machinelog WHERE machineId = %s", (self.id,)) + q = """DELETE FROM machines WHERE id=%s""" + cur.execute(q, (self.id,)) + get_db().commit() @staticmethod - def inRange(r, ip): - ''' + def inRange(r: str, ip: str | IPAddress) -> bool: + """ returns true if ip is in range r, false otherwise. @param r: range @param ip: ip address @return true if ip is in range r, false otherwise. - ''' - #single ip? - m = re.search(r'^[\d]{1-3}\.[\d]{1-3}\.[\d]{1-3}\.[\d]{1-3}$', r) + """ + # single ip? + m = re.search(r"^[\d]{1-3}\.[\d]{1-3}\.[\d]{1-3}\.[\d]{1-3}$", r) if m: ri = IPAddress(r) return ri == r - #hostname? - m = re.search(r'^[^\d][^ ]+$', r) + # hostname? + m = re.search(r"^[^\d][^ ]+$", r) if m: return socket.gethostbyname(r) == str(ip) - #cidr range as string? + # cidr range as string? try: rn = IPNetwork(r) return IPAddress(ip) in rn - except: - raise Exception('cannot determine if %s is in range %s' % (ip, r)) + except (AddrFormatError, ValueError): + raise Exception("cannot determine if %s is in range %s" % (ip, r)) @staticmethod def getIPsInRange(r, exposedOnly=False): - ''' + """ return a dictionary of IPs to IDs for all known hosts in the defined range. @param r: ip range @return dictionary IPs/Machine IDs in that range - ''' + """ - cur = request.db.cursor() + cur = get_db().cursor() q = """select id, ip from machines""" cur.execute(q) @@ -1041,8 +1077,8 @@ def getIPsInRange(r, exposedOnly=False): return inRangeIPs -def scan(target, port_range, extra_options, job_id, sensor='localhost'): - ''' +def scan(target, port_range, extra_options, job_id, sensor="localhost"): + """ Scans the port range on the target IP/IP Range with nmap. extra_options are passed as CLI arguments to nmap. The job_id will be saved to the changelog entry in the database @@ -1054,152 +1090,144 @@ def scan(target, port_range, extra_options, job_id, sensor='localhost'): @param job_id job UUID as used by the schedule model @param sensor In future versions, you may specify a sensor to be used for scanning (not implemented yet). - ''' - #get db - cur = request.db.cursor() + """ + # get db + cur = get_db().cursor() - #set all IN PROGRESS log entries older than one day to TIMEOUT + # set all IN PROGRESS log entries older than one day to TIMEOUT q = """update scanlog set state='TIMEOUT' where state='IN PROGRESS' and datediff(now(), startdate) > 1""" cur.execute(q) - #log start of scan + # log start of scan q = """INSERT INTO scanlog (jobid, state, startdate, iprange, portrange, extraoptions) - VALUES ("%s", "%s", NOW(), "%s", %s, %s)""" % ( - "%s" % job_id if job_id else "Null", "IN PROGRESS", - target, - "%s" % port_range if port_range else "Null", - "'%s'" % extra_options if extra_options else "Null") - - cur.execute(q) + VALUES (%s, %s, NOW(), %s, %s, %s)""" + cur.execute( + q, + ( + job_id if job_id else None, + "IN PROGRESS", + target, + port_range if port_range else None, + extra_options if extra_options else None, + ), + ) logid = cur.lastrowid - request.db.commit() + get_db().commit() cur.close() try: - - #start port scanner + # start port scanner ps = nmap.PortScanner() - #recode to ascii - utf not allowed - ps.scan(anyascii(target), anyascii(port_range) if - port_range else None, anyascii(extra_options)) - - cur = request.db.cursor() - - #machines - #remove old machines - oldmachs = Machine.getIPsInRange(anyascii(target), - exposedOnly=True) - print("known machines in range %s: %s" % (target, oldmachs)) - print("machines found by scan: %s" % ps.all_hosts()) - for hostip in [x for x in list(oldmachs.keys()) if (False if x in ps.all_hosts() else True)]: + # recode to ascii - utf not allowed + ps.scan( + anyascii(target), anyascii(port_range) if port_range else None, anyascii(extra_options) + ) + + cur = get_db().cursor() + + # machines + # remove old machines + oldmachs = Machine.getIPsInRange(anyascii(target), exposedOnly=True) + logger.debug("known machines in range", target=target, machines=oldmachs) + logger.debug("machines found by scan", hosts=ps.all_hosts()) + for hostip in [ + x for x in list(oldmachs.keys()) if (False if x in ps.all_hosts() else True) + ]: mach = Machine(oldmachs[hostip]) for svc in mach.getServices(exposedOnly=True): - print("deleting service %s on %s" % (svc.port, hostip)) + logger.info("deleting service", port=svc.port, host=hostip) svc.delete() - #create new machines, add/remove services + # create new machines, add/remove services for hostip in ps.all_hosts(): host = ps[hostip] - #if nmap did not determine hostname, try be reverse name resolution - if not ('hostname' in host and host['hostname'] and len(host['hostname'])): + # if nmap did not determine hostname, try be reverse name resolution + if not ("hostname" in host and host["hostname"] and len(host["hostname"])): from socket import gethostbyaddr, herror + try: - host['hostname'] = gethostbyaddr(hostip)[0] + host["hostname"] = gethostbyaddr(hostip)[0] except herror: - host['hostname'] = '' - - #create new machines - mach = Machine.create(host['hostname'], hostip) - #open ports - if 'tcp' in host: - portnumbers = [x for x in list(host['tcp'].keys()) if (True if host['tcp'][x]['state'] - == 'open' else False)] - #delete old services + host["hostname"] = "" + + # create new machines + mach = Machine.create(host["hostname"], hostip) + # open ports + if "tcp" in host: + portnumbers = [ + x + for x in list(host["tcp"].keys()) + if (True if host["tcp"][x]["state"] == "open" else False) + ] + # delete old services for svc in mach.getServices(exposedOnly=True): # if port not detected anymore and in scanned range, delete # it - if not svc.port in portnumbers and (svc.inRange(port_range - ) if port_range else True): - print("deleting %s" % svc.port) + if svc.port not in portnumbers and ( + svc.inRange(port_range) if port_range else True + ): + logger.info("deleting service", port=svc.port) svc.delete() - #create new services + # create new services for portno in portnumbers: - port = host['tcp'][portno] - if port['state'] == 'open': - print("creating %s" % portno) - Service.create(mach, portno, port['product'] if - "product" in port else None, port['version'] if - "version" in port else None, port['extrainfo'] if - "extrainfo" in port else None) - - #log + port = host["tcp"][portno] + if port["state"] == "open": + logger.info("creating service", port=portno) + Service.create( + mach, + portno, + port["product"] if "product" in port else None, + port["version"] if "version" in port else None, + port["extrainfo"] if "extrainfo" in port else None, + ) + + # log state = "OK" message = None - except: + except Exception: state = "FAILED" message = traceback.format_exc() - print(message) + logger.exception("scan failed") finally: - #log - cur = request.db.cursor() - q = """UPDATE scanlog SET state="%s", enddate=NOW(), output=%s - WHERE ID=%s""" + # log + cur = get_db().cursor() + q = "UPDATE scanlog SET state=%s, enddate=NOW(), output=%s WHERE id=%s" + cur.execute(q, (state, message, logid)) - cur.execute(q, (state, "'%s'" % request.db.escape_string(message) if message - else "NULL", logid)) - - request.db.commit() + get_db().commit() return state -def initDB(localconf): - ''' - inits the database into the bottle request scope. - ''' - try: - request.cfg = localconf - except AttributeError as e: - pass - try: - request.db = mdb.connect(**request.cfg.db.data) - except AttributeError as e: - pass - -def closeDB(): - ''' - commit all open database cursors. close the connection. - ''' - try: - request.db.commit() - request.db.close() - except request.db.OperationalError: - pass - - -def genMessages(exp): - ''' +def genMessages(exp: list[dict[str, Any]]) -> tuple[list[str], list[str]]: + """ generates textual descriptions of profile discrepancies. @param rowset: a rowset as generated by aspromModel.getAlertsExposed() or getAlertsClosed() @return a two-tuple containing a list of critical and a list of\ warning discrepancies - ''' + """ messageCrit = [] messageWarn = [] for row in exp: - alertFmt = (("%s[%s]" % (row['service'], str(row['port']))) if row[ - 'service'] else str(row['port'])) + ' on ' + (row['hostname'] if - row['hostname'] else row['ip']) - if row['crit']: + alertFmt = ( + ( + ("%s[%s]" % (row["service"], str(row["port"]))) + if row["service"] + else str(row["port"]) + ) + + " on " + + (row["hostname"] if row["hostname"] else row["ip"]) + ) + if row["crit"]: messageCrit.append(alertFmt) else: messageWarn.append(alertFmt) diff --git a/inc/db.py b/inc/db.py new file mode 100644 index 0000000..2e9ee10 --- /dev/null +++ b/inc/db.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import threading +from typing import TYPE_CHECKING, Any + +import MySQLdb as mdb +from bottle import request + +if TYPE_CHECKING: + from inc.asprom import Cfg + +_local = threading.local() + + +def set_db(conn: Any) -> None: + """Set the active connection for non-Bottle contexts (tests, CLI).""" + _local.connection = conn + + +def set_cfg(cfg: Cfg) -> None: + """Set the active configuration for non-Bottle contexts.""" + _local.cfg = cfg + + +def get_db() -> Any: + """Return the active database connection.""" + try: + return request.db + except (AttributeError, RuntimeError): + conn = getattr(_local, "connection", None) + if conn is None: + raise RuntimeError("No database connection available") + return conn + + +def get_cfg() -> Cfg: + """Return the active configuration object.""" + try: + return request.cfg + except (AttributeError, RuntimeError): + cfg = getattr(_local, "cfg", None) + if cfg is None: + raise RuntimeError("No configuration available") + return cfg + + +def init_db(localconf: Cfg) -> None: + """Open a database connection and bind it to the current context.""" + try: + request.cfg = localconf + request.db = mdb.connect(**localconf.db.data) + except (AttributeError, RuntimeError): + set_cfg(localconf) + set_db(mdb.connect(**localconf.db.data)) + + +def close_db() -> None: + """Commit and close the active database connection.""" + try: + db = get_db() + db.commit() + db.close() + except mdb.OperationalError: + pass + except RuntimeError: + pass + finally: + if hasattr(_local, "connection"): + del _local.connection + if hasattr(_local, "cfg"): + del _local.cfg + try: + del request.db + except (AttributeError, RuntimeError): + pass diff --git a/inc/logging.py b/inc/logging.py new file mode 100644 index 0000000..7ad1511 --- /dev/null +++ b/inc/logging.py @@ -0,0 +1,39 @@ +"""Structured logging configuration for asprom.""" + +from __future__ import annotations + +import logging +import os +import sys + +import structlog + + +def configure_logging(json_output: bool | None = None) -> None: + """Configure structlog for console or JSON output.""" + if json_output is None: + json_output = os.environ.get("ASPROM_LOG_FORMAT", "").lower() == "json" + + processors = [ + structlog.contextvars.merge_contextvars, + structlog.processors.add_log_level, + structlog.processors.TimeStamper(fmt="iso"), + structlog.processors.StackInfoRenderer(), + ] + if json_output: + processors.append(structlog.processors.JSONRenderer()) + else: + processors.append(structlog.dev.ConsoleRenderer()) + + structlog.configure( + processors=processors, # type: ignore[arg-type] + wrapper_class=structlog.make_filtering_bound_logger(logging.INFO), + context_class=dict, + logger_factory=structlog.PrintLoggerFactory(file=sys.stderr), + cache_logger_on_first_use=True, + ) + + +def get_logger(name: str, **initial: object) -> structlog.BoundLogger: + """Return a bound structlog logger.""" + return structlog.get_logger(name).bind(**initial) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..8661b79 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,62 @@ +[project] +name = "asprom" +version = "0.1.0" +description = "Assault Profile Monitor - network security compliance scanner" +readme = "README.md" +requires-python = ">=3.11" +dependencies = [ + "anyascii==0.3.2", + "mysqlclient==2.2.4", + "python-crontab==3.2.0", + "netaddr==1.3.0", + "paste==3.10.1", + "bottle==0.13.1", + "config==0.4.2", + "croniter==3.0.3", + "prometheus-client==0.20.0", + "python-nmap==0.7.1", + "structlog>=24.0.0", + "alembic>=1.13.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0.0", + "pytest-cov>=5.0.0", + "testcontainers[mysql]>=4.0.0", + "ruff>=0.4.0", + "mypy>=1.10.0", +] + +[tool.ruff] +target-version = "py311" +line-length = 100 +src = [".", "inc", "tests"] +exclude = ["docker/patch-crontab.py"] + +[tool.ruff.lint] +select = ["E", "F", "W", "I"] +ignore = ["E501", "E722"] + +[tool.ruff.lint.per-file-ignores] +"aspromNagiosCheck.py" = ["F821"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["."] + +[tool.mypy] +python_version = "3.11" +ignore_missing_imports = true +warn_return_any = false +warn_unused_ignores = true + +[tool.coverage.run] +source = ["inc"] +omit = ["tests/*"] + +[tool.coverage.report] +exclude_lines = [ + "pragma: no cover", + "if TYPE_CHECKING:", +] diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..3f77109 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,6 @@ +[pytest] +testpaths = tests +pythonpath = . +markers = + integration: tests requiring Docker/testcontainers MySQL + xfail: known failures pending fix diff --git a/requirements.txt b/requirements.txt index 5aab991..24bbda5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ anyascii==0.3.2 -mysqlclient==2.2.4 +mysqlclient==2.2.8 python-crontab==3.2.0 netaddr==1.3.0 paste==3.10.1 @@ -8,3 +8,6 @@ config==0.4.2 croniter==3.0.3 prometheus-client==0.20.0 python-nmap==0.7.1 +structlog>=24.0.0 +alembic>=1.13.0 +sqlalchemy>=2.0.0 diff --git a/scripts/check.sh b/scripts/check.sh new file mode 100755 index 0000000..27bab65 --- /dev/null +++ b/scripts/check.sh @@ -0,0 +1,20 @@ +#!/usr/bin/env bash +set -euo pipefail + +cd "$(dirname "$0")/.." + +ruff check . +ruff format --check . + +if [ -d tests ] && [ -n "$(find tests -name 'test_*.py' -print -quit 2>/dev/null)" ]; then + COV_ARGS=(--cov=inc --cov-report=term-missing) + if [ -n "${ASPROM_COV_FAIL_UNDER:-}" ]; then + COV_ARGS+=(--cov-fail-under="${ASPROM_COV_FAIL_UNDER}") + fi + pytest "${COV_ARGS[@]}" + if [ "${ASPROM_RUN_MYPY:-1}" = "1" ]; then + mypy inc/ + fi +else + echo "No tests yet — skipping pytest" +fi diff --git a/scripts/dev-prod.sh b/scripts/dev-prod.sh new file mode 100755 index 0000000..ab725b9 --- /dev/null +++ b/scripts/dev-prod.sh @@ -0,0 +1,246 @@ +#!/usr/bin/env bash +# Run aspromGUI locally against the production MySQL database in Kubernetes. +# +# Discovers the MySQL service name via kubectl, port-forwards it to localhost, +# and starts the Bottle GUI pointed at the forwarded connection. +# +# Prerequisites: +# - kubectl configured with access to the target cluster +# - Python venv with deps (./scripts/setup-venv.sh && source venv/bin/activate) +# +# Usage: +# ./scripts/dev-prod.sh +# +# Override defaults: +# KUBE_CONTEXT=internal1 KUBE_NAMESPACE=asprom LOCAL_MYSQL_PORT=3307 ./scripts/dev-prod.sh +# +# WARNING: You are connecting to the production database. Scans, baseline +# changes, and deletions affect live data. + +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "$ROOT" + +if [ -x "$ROOT/venv/bin/python3" ]; then + PYTHON="$ROOT/venv/bin/python3" +else + PYTHON="${PYTHON:-python3}" +fi + +KUBE_CONTEXT="${KUBE_CONTEXT:-internal1}" +KUBE_NAMESPACE="${KUBE_NAMESPACE:-asprom}" +LOCAL_MYSQL_PORT="${LOCAL_MYSQL_PORT:-3307}" +GUI_PORT="${GUI_PORT:-8080}" +MYSQL_SERVICE="${MYSQL_SERVICE:-}" +DB_USER="${ASPROM_DB_USER:-asprom}" +DB_NAME="${ASPROM_DB_NAME:-asprom}" +DB_PASSWORD="${ASPROM_DB_PASSWORD:-}" + +PF_PID="" +TMP_CFG="" +cleanup() { + if [ -n "$PF_PID" ] && kill -0 "$PF_PID" 2>/dev/null; then + kill "$PF_PID" 2>/dev/null || true + wait "$PF_PID" 2>/dev/null || true + fi + if [ -n "$TMP_CFG" ] && [ -f "$TMP_CFG" ]; then + rm -f "$TMP_CFG" + fi +} +trap cleanup EXIT INT TERM + +require_cmd() { + if ! command -v "$1" >/dev/null 2>&1; then + echo "error: required command not found: $1" >&2 + exit 1 + fi +} + +kubectl_ctx() { + kubectl --context "$KUBE_CONTEXT" -n "$KUBE_NAMESPACE" "$@" +} + +discover_mysql_service() { + if [ -n "$MYSQL_SERVICE" ]; then + echo "$MYSQL_SERVICE" + return + fi + + if kubectl_ctx get svc mysql >/dev/null 2>&1; then + echo mysql + return + fi + + local mysql_named + mysql_named="$(kubectl_ctx get svc -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}' \ + | grep -i mysql | head -1 || true)" + if [ -n "$mysql_named" ]; then + echo "$mysql_named" + return + fi + + local port_match + port_match="$(kubectl_ctx get svc -o json \ + | python3 -c " +import json, sys +data = json.load(sys.stdin) +matches = [] +for item in data.get('items', []): + for port in item.get('spec', {}).get('ports', []): + if port.get('port') == 3306: + matches.append(item['metadata']['name']) +if len(matches) == 1: + print(matches[0]) +elif len(matches) > 1: + print('MULTIPLE:' + ','.join(matches), file=sys.stderr) + sys.exit(2) +")" + + if [ -n "$port_match" ]; then + echo "$port_match" + return + fi + + echo "error: could not discover a MySQL service in context=$KUBE_CONTEXT namespace=$KUBE_NAMESPACE" >&2 + echo "Services in namespace:" >&2 + kubectl_ctx get svc >&2 || true + echo "Set MYSQL_SERVICE explicitly, e.g. MYSQL_SERVICE=mysql ./scripts/dev-prod.sh" >&2 + exit 1 +} + +discover_mysql_port() { + local svc="$1" + kubectl_ctx get svc "$svc" -o jsonpath='{.spec.ports[?(@.port==3306)].port}' +} + +discover_db_password() { + if [ -n "$DB_PASSWORD" ]; then + echo "$DB_PASSWORD" + return + fi + + local secret_keys secret_name value + for secret_name in mysql asprom-mysql asprom; do + if ! kubectl_ctx get secret "$secret_name" >/dev/null 2>&1; then + continue + fi + for secret_keys in password mysql-password MYSQL_PASSWORD; do + value="$(kubectl_ctx get secret "$secret_name" -o "jsonpath={.data.${secret_keys}}" 2>/dev/null || true)" + if [ -n "$value" ]; then + echo "$value" | base64 -d + return + fi + done + done + + value="$(kubectl_ctx get deploy -o json \ + | python3 -c " +import base64, json, sys +data = json.load(sys.stdin) +keys = ('MYSQL_PASSWORD', 'DB_PASSWORD', 'password') +for item in data.get('items', []): + for container in item.get('spec', {}).get('template', {}).get('spec', {}).get('containers', []): + for env in container.get('env', []): + name = env.get('name', '') + if name in keys and env.get('value'): + print(env['value']) + sys.exit(0) + ref = env.get('valueFrom', {}).get('secretKeyRef', {}) + if ref.get('key') in keys: + print(f\"SECRET:{ref.get('name')}:{ref.get('key')}\") + sys.exit(0) +" 2>/dev/null || true)" + + if [[ "$value" == SECRET:* ]]; then + IFS=: read -r _ secret_name secret_key <<<"$value" + value="$(kubectl_ctx get secret "$secret_name" -o "jsonpath={.data.${secret_key}}" | base64 -d)" + echo "$value" + return + fi + + if [ -n "$value" ]; then + echo "$value" + return + fi + + echo "error: could not discover database password; set ASPROM_DB_PASSWORD" >&2 + exit 1 +} + +write_local_config() { + local port="$1" + TMP_CFG="$(mktemp "${TMPDIR:-/tmp}/asprom-prod-local.XXXXXX.cfg")" + cat >"$TMP_CFG" </dev/null 2>&1; then + echo "error: Python not found ($PYTHON). Run ./scripts/setup-venv.sh first." >&2 + exit 1 + fi + + echo "Discovering MySQL service (context=$KUBE_CONTEXT, namespace=$KUBE_NAMESPACE)..." + MYSQL_SERVICE="$(discover_mysql_service)" + REMOTE_PORT="$(discover_mysql_port "$MYSQL_SERVICE")" + if [ -z "$REMOTE_PORT" ]; then + REMOTE_PORT=3306 + fi + + DB_PASSWORD="$(discover_db_password)" + + echo "MySQL service: $MYSQL_SERVICE (remote port $REMOTE_PORT -> localhost:$LOCAL_MYSQL_PORT)" + echo "Database: $DB_USER@127.0.0.1:$LOCAL_MYSQL_PORT/$DB_NAME" + echo + echo "WARNING: connected to production data. Press Ctrl+C to stop." + echo + + kubectl_ctx port-forward "svc/$MYSQL_SERVICE" "${LOCAL_MYSQL_PORT}:${REMOTE_PORT}" >/dev/null & + PF_PID=$! + + wait_for_mysql "$LOCAL_MYSQL_PORT" + write_local_config "$LOCAL_MYSQL_PORT" + + exec "$PYTHON" aspromGUI.py +} + +main "$@" diff --git a/scripts/setup-venv.sh b/scripts/setup-venv.sh new file mode 100755 index 0000000..595d743 --- /dev/null +++ b/scripts/setup-venv.sh @@ -0,0 +1,66 @@ +#!/usr/bin/env bash +# Create a local venv and install asprom Python dependencies. +# +# Debian/Ubuntu (PEP 668): use a venv; mysqlclient also needs MySQL client dev headers. +# +# Usage: +# ./scripts/setup-venv.sh +# source venv/bin/activate +# ./scripts/dev-prod.sh + +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "$ROOT" + +PYTHON="${PYTHON:-}" +if [ -z "$PYTHON" ]; then + for candidate in python3.12 python3.11 python3; do + if command -v "$candidate" >/dev/null 2>&1; then + PYTHON="$candidate" + break + fi + done +fi + +if [ -z "$PYTHON" ]; then + echo "error: no python3 interpreter found" >&2 + exit 1 +fi + +if ! "$PYTHON" -c "import sys; sys.exit(0 if sys.version_info >= (3, 10) else 1)"; then + echo "error: $PYTHON must be Python 3.10 or newer" >&2 + exit 1 +fi + +missing_pkgs=() +for pkg in default-libmysqlclient-dev pkg-config build-essential; do + if ! dpkg -s "$pkg" >/dev/null 2>&1; then + missing_pkgs+=("$pkg") + fi +done + +if [ "${#missing_pkgs[@]}" -gt 0 ]; then + echo "Missing system packages required to build mysqlclient:" >&2 + printf ' %s\n' "${missing_pkgs[@]}" >&2 + echo >&2 + echo "Install on Debian/Ubuntu:" >&2 + echo " sudo apt install -y python3-venv ${missing_pkgs[*]} nmap" >&2 + exit 1 +fi + +if [ ! -d venv ]; then + echo "Creating venv with $PYTHON ..." + "$PYTHON" -m venv venv +fi + +echo "Installing Python dependencies into venv ..." +venv/bin/pip install --upgrade pip +venv/bin/pip install -r requirements.txt + +echo +echo "Done. Activate the venv with:" +echo " source venv/bin/activate" +echo +echo "Then run:" +echo " ./scripts/dev-prod.sh" diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..0d3eb96 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,184 @@ +"""Pytest fixtures for asprom integration tests.""" + +from __future__ import annotations + +import os +import re +from pathlib import Path + +import MySQLdb +import pytest + +ROOT = Path(__file__).resolve().parent.parent +DDL_PATH = ROOT / "db" / "ddl.sql" + + +class DbTestCfg: + """Minimal configuration for test database connections.""" + + def __init__(self, host: str, port: int, user: str, password: str, database: str): + self.maindir = str(ROOT) + self.db = type( + "DbSection", + (), + { + "data": { + "host": host, + "port": port, + "user": user, + "passwd": password, + "db": database, + } + }, + )() + self.misc = type("MiscSection", (), {"url": "http://localhost:8080"})() + + +def _load_schema(connection: MySQLdb.Connection) -> None: + sql = DDL_PATH.read_text(encoding="utf-8") + # Unwrap MySQL conditional comments (preserve VIEW definitions) + sql = re.sub(r"/\*!50001\s+(.*?)\s*\*/", r"\1", sql, flags=re.DOTALL) + sql = re.sub(r"/\*![0-9]+\s+(.*?)\s*\*/", r"\1", sql, flags=re.DOTALL) + sql = re.sub(r"/\*![0-9]+.*?\*/", "", sql, flags=re.DOTALL) + statements = [s.strip() for s in sql.split(";") if s.strip()] + cur = connection.cursor() + cur.execute("SET FOREIGN_KEY_CHECKS = 0") + for statement in statements: + upper = statement.upper() + if ( + upper.startswith("USE ") + or upper.startswith("SET @") + or "CHARACTER_SET" in upper + or "COLLATION_CONNECTION" in upper + or "SQL_MODE" in upper + or "FOREIGN_KEY_CHECKS=@OLD" in upper + or "UNIQUE_CHECKS=@OLD" in upper + or "TIME_ZONE=@OLD" in upper + or "SQL_NOTES=@OLD" in upper + ): + continue + try: + cur.execute(statement) + except MySQLdb.Error as exc: + if exc.args[0] not in (1050, 1051): + raise + cur.execute("SET FOREIGN_KEY_CHECKS = 1") + connection.commit() + + +def _default_params() -> dict: + return { + "host": os.environ.get("ASPROM_TEST_DB_HOST", "127.0.0.1"), + "port": int(os.environ.get("ASPROM_TEST_DB_PORT", "3306")), + "user": os.environ.get("ASPROM_TEST_DB_USER", "asprom"), + "passwd": os.environ.get("ASPROM_TEST_DB_PASSWORD", "asprom"), + "db": os.environ.get("ASPROM_TEST_DB_NAME", "asprom_test"), + } + + +@pytest.fixture(scope="session") +def mysql_params(): + params = _default_params() + use_testcontainers = os.environ.get("ASPROM_USE_TESTCONTAINERS", "").lower() in ( + "1", + "true", + "yes", + ) + container = None + + if use_testcontainers: + from testcontainers.mysql import MySqlContainer + + container = MySqlContainer("mysql:8.0") + container.start() + params = { + "host": container.get_container_host_ip(), + "port": int(container.get_exposed_port(3306)), + "user": container.username, + "passwd": container.password, + "db": container.dbname, + } + + try: + conn = MySQLdb.connect(**params) + _load_schema(conn) + conn.close() + except MySQLdb.Error as exc: + if container is not None: + container.stop() + pytest.skip(f"MySQL not available for integration tests: {exc}") + + yield params + + if container is not None: + container.stop() + + +@pytest.fixture +def db_connection(mysql_params): + from inc.db import close_db, set_cfg, set_db + + conn = MySQLdb.connect( + host=mysql_params["host"], + port=mysql_params["port"], + user=mysql_params["user"], + passwd=mysql_params["passwd"], + db=mysql_params["db"], + ) + _truncate_tables(conn) + cfg = DbTestCfg( + host=mysql_params["host"], + port=mysql_params["port"], + user=mysql_params["user"], + password=mysql_params["passwd"], + database=mysql_params["db"], + ) + set_cfg(cfg) + set_db(conn) + yield conn + close_db() + + +def _truncate_tables(conn: MySQLdb.Connection) -> None: + cur = conn.cursor() + cur.execute("SET FOREIGN_KEY_CHECKS = 0") + for table in ( + "changelog", + "criticality", + "machinelog", + "scanlog", + "servicelog", + "services", + "machines", + ): + cur.execute(f"TRUNCATE TABLE `{table}`") + cur.execute("SET FOREIGN_KEY_CHECKS = 1") + conn.commit() + + +@pytest.fixture +def seed_machine(db_connection): + from inc.asprom import Machine + + def _seed(name: str = "host1", ip: str = "10.0.0.1") -> Machine: + return Machine.create(name, ip) + + return _seed + + +@pytest.fixture +def seed_service(db_connection, seed_machine): + from inc.asprom import Service + + def _seed(port: int = 22, product: str = "ssh", machine=None) -> Service: + mach = machine or seed_machine() + return Service.create(mach, port, product=product) + + return _seed + + +@pytest.fixture +def asprom_model(db_connection): + from inc.asprom import AspromModel + + return AspromModel(username="testuser") diff --git a/tests/helpers.py b/tests/helpers.py new file mode 100644 index 0000000..051dc6f --- /dev/null +++ b/tests/helpers.py @@ -0,0 +1,55 @@ +"""Shared test assertions for asprom domain model.""" + +from __future__ import annotations + +import MySQLdb + + +def assert_in_exposed(db: MySQLdb.Connection, service_id: int) -> None: + cur = db.cursor() + cur.execute("SELECT COUNT(*) FROM exposed WHERE id = %s", (service_id,)) + count = cur.fetchone()[0] + assert count == 1, f"service {service_id} not in exposed view" + + +def assert_not_in_exposed(db: MySQLdb.Connection, service_id: int) -> None: + cur = db.cursor() + cur.execute("SELECT COUNT(*) FROM exposed WHERE id = %s", (service_id,)) + count = cur.fetchone()[0] + assert count == 0, f"service {service_id} still in exposed view" + + +def assert_in_baseline(db: MySQLdb.Connection, service_id: int) -> None: + cur = db.cursor() + cur.execute( + "SELECT COUNT(*) FROM neatline WHERE serviceId = %s AND neat = 1", + (service_id,), + ) + count = cur.fetchone()[0] + assert count == 1, f"service {service_id} not in baseline" + + +def assert_not_in_baseline(db: MySQLdb.Connection, service_id: int) -> None: + cur = db.cursor() + cur.execute( + "SELECT neat FROM neatline WHERE serviceId = %s ORDER BY id DESC LIMIT 1", + (service_id,), + ) + row = cur.fetchone() + assert row is None or row[0] == 0, f"service {service_id} still in baseline" + + +def count_servicelog_entries(db: MySQLdb.Connection, service_id: int) -> int: + cur = db.cursor() + cur.execute("SELECT COUNT(*) FROM servicelog WHERE serviceId = %s", (service_id,)) + return cur.fetchone()[0] + + +def current_service_open(db: MySQLdb.Connection, service_id: int) -> bool: + cur = db.cursor() + cur.execute( + "SELECT openp FROM servicelogCur WHERE serviceId = %s", + (service_id,), + ) + row = cur.fetchone() + return bool(row[0]) if row else False diff --git a/tests/test_baseline.py b/tests/test_baseline.py new file mode 100644 index 0000000..c7282e0 --- /dev/null +++ b/tests/test_baseline.py @@ -0,0 +1,44 @@ +"""Integration tests for baseline approve/remove workflow.""" + +from tests.helpers import ( + assert_in_baseline, + assert_in_exposed, + assert_not_in_baseline, + assert_not_in_exposed, +) + + +def test_approve_moves_to_baseline(seed_service, db_connection, asprom_model): + svc = seed_service(port=2222, product="ssh") + assert_in_exposed(db_connection, svc.id) + + svc.approve("business need", "testuser") + assert_not_in_exposed(db_connection, svc.id) + assert_in_baseline(db_connection, svc.id) + + cur = db_connection.cursor() + cur.execute( + "SELECT justification, username FROM changelog WHERE serviceId = %s ORDER BY id DESC LIMIT 1", + (svc.id,), + ) + row = cur.fetchone() + assert row[0] == "business need" + assert row[1] == "testuser" + + +def test_remove_from_baseline(seed_service, db_connection): + svc = seed_service(port=3333) + svc.approve("approved", "admin") + assert_in_baseline(db_connection, svc.id) + + svc.remove("no longer needed", "admin") + assert_not_in_baseline(db_connection, svc.id) + assert_in_exposed(db_connection, svc.id) + + +def test_get_neatline_after_approve(seed_service, asprom_model): + svc = seed_service(port=4444, product="mysql") + svc.approve("db server", "ops") + rows = asprom_model.getNeatline() + ids = [r["id"] for r in rows] + assert svc.id in ids diff --git a/tests/test_coverage_gaps.py b/tests/test_coverage_gaps.py new file mode 100644 index 0000000..50ca682 --- /dev/null +++ b/tests/test_coverage_gaps.py @@ -0,0 +1,33 @@ +"""Additional coverage for metrics and controller paths.""" + +from unittest.mock import patch + +import aspromMetrics + + +def test_refresh_metrics_sets_gauges(): + with patch.object(aspromMetrics.alertsExposed, "set") as mock_exposed: + with patch.object(aspromMetrics.alertsClosed, "set") as mock_closed: + with patch.object(aspromMetrics, "_ensure_model") as mock_model: + mock_model.return_value.getAlertsExposed.return_value = [{"id": 1}] + mock_model.return_value.getAlertsClosed.return_value = [ + {"id": 2}, + {"id": 3}, + ] + aspromMetrics.refreshMetrics() + mock_exposed.assert_called_once_with(1) + mock_closed.assert_called_once_with(2) + + +@patch("inc.asprom.scan", return_value="OK") +def test_controller_rescan_job(mock_scan): + from inc.asprom import Controller + + with patch("inc.asprom.AspromScheduleModel") as mock_sm: + instance = mock_sm.return_value + instance.promoteToIndex.return_value = { + "job-1": {"iprange": "10.0.0.0/24", "ports": "22", "params": ""} + } + result = Controller.rescanJob("job-1") + assert result == "OK" + mock_scan.assert_called_once() diff --git a/tests/test_db_context.py b/tests/test_db_context.py new file mode 100644 index 0000000..4cc2da0 --- /dev/null +++ b/tests/test_db_context.py @@ -0,0 +1,24 @@ +"""Tests for inc.db connection context.""" + +import MySQLdb + +from inc.db import close_db, get_db, set_cfg, set_db +from tests.conftest import DbTestCfg + + +def test_get_db_outside_bottle(db_connection, mysql_params): + cfg = DbTestCfg( + host=mysql_params["host"], + port=mysql_params["port"], + user=mysql_params["user"], + password=mysql_params["passwd"], + database=mysql_params["db"], + ) + conn = MySQLdb.connect(**mysql_params) + set_cfg(cfg) + set_db(conn) + db = get_db() + cur = db.cursor() + cur.execute("SELECT 1") + assert cur.fetchone()[0] == 1 + close_db() diff --git a/tests/test_gen_messages.py b/tests/test_gen_messages.py new file mode 100644 index 0000000..27033e3 --- /dev/null +++ b/tests/test_gen_messages.py @@ -0,0 +1,38 @@ +"""Tests for genMessages alert formatting.""" + +from inc.asprom import genMessages + + +def test_gen_messages_empty(): + crit, warn = genMessages([]) + assert crit == [] + assert warn == [] + + +def test_gen_messages_critical_only(): + rows = [ + {"service": "ssh", "port": 22, "hostname": "host1", "ip": "10.0.0.1", "crit": True}, + ] + crit, warn = genMessages(rows) + assert crit == ["ssh[22] on host1"] + assert warn == [] + + +def test_gen_messages_warning_only(): + rows = [ + {"service": "http", "port": 80, "hostname": "", "ip": "10.0.0.2", "crit": False}, + ] + crit, warn = genMessages(rows) + assert crit == [] + assert warn == ["http[80] on 10.0.0.2"] + + +def test_gen_messages_mixed(): + rows = [ + {"service": "ssh", "port": 22, "hostname": "a", "ip": "10.0.0.1", "crit": True}, + {"service": "", "port": 443, "hostname": "", "ip": "10.0.0.2", "crit": False}, + ] + crit, warn = genMessages(rows) + assert len(crit) == 1 + assert len(warn) == 1 + assert "443" in warn[0] diff --git a/tests/test_in_range.py b/tests/test_in_range.py new file mode 100644 index 0000000..6b96d00 --- /dev/null +++ b/tests/test_in_range.py @@ -0,0 +1,40 @@ +"""Tests for Service.inRange and Machine.inRange.""" + +import pytest + +from inc.asprom import Machine, Service + + +class TestServiceInRange: + def test_single_port_int(self): + svc = type("S", (), {"port": 22})() + assert Service.inRange(svc, 22) is True + assert Service.inRange(svc, 80) is False + + def test_single_port_string(self): + svc = type("S", (), {"port": 443})() + assert Service.inRange(svc, "443") is True + + def test_port_range(self): + svc = type("S", (), {"port": 8080})() + assert Service.inRange(svc, "8000-9000") is True + assert Service.inRange(svc, "1-80") is False + + def test_invalid_range_raises(self): + svc = type("S", (), {"port": 22})() + with pytest.raises(Exception): + Service.inRange(svc, "invalid") + + +class TestMachineInRange: + def test_single_ip(self): + assert Machine.inRange("10.0.0.1", "10.0.0.1") is True + assert Machine.inRange("10.0.0.2", "10.0.0.1") is False + + def test_cidr_range(self): + assert Machine.inRange("10.0.0.0/24", "10.0.0.50") is True + assert Machine.inRange("10.0.0.0/24", "10.0.1.1") is False + + def test_invalid_range_raises(self): + with pytest.raises(Exception): + Machine.inRange("not-a-range", "10.0.0.1") diff --git a/tests/test_machine.py b/tests/test_machine.py new file mode 100644 index 0000000..8e60088 --- /dev/null +++ b/tests/test_machine.py @@ -0,0 +1,31 @@ +"""Integration tests for Machine model.""" + +from inc.asprom import Machine + + +def test_machine_create_upsert_by_ip(db_connection): + m1 = Machine.create("host-a", "10.1.1.1") + m2 = Machine.create("host-a-renamed", "10.1.1.1") + assert m1.id == m2.id + assert m2.hostname == "host-a-renamed" + + +def test_get_services_exposed_only(seed_service, db_connection): + mach = seed_service(port=22).machine + open_svc = seed_service(port=80, machine=mach) + open_svc.delete() + + exposed = mach.getServices(exposedOnly=True) + assert all(s.port == 22 for s in exposed) + + +def test_delete_machine(seed_service, db_connection): + mach = seed_service(port=9000).machine + machine_id = mach.id + for service in mach.getServices(): + service.delete() + mach.delete() + + cur = db_connection.cursor() + cur.execute("SELECT COUNT(*) FROM machines WHERE id = %s", (machine_id,)) + assert cur.fetchone()[0] == 0 diff --git a/tests/test_migrations.py b/tests/test_migrations.py new file mode 100644 index 0000000..2c77e52 --- /dev/null +++ b/tests/test_migrations.py @@ -0,0 +1,32 @@ +"""Alembic migration tests.""" + +import os +import subprocess +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parent.parent + + +@pytest.mark.integration +def test_alembic_upgrade_head(mysql_params): + env = os.environ.copy() + env.update( + { + "ASPROM_TEST_DB_HOST": mysql_params["host"], + "ASPROM_TEST_DB_PORT": str(mysql_params["port"]), + "ASPROM_TEST_DB_USER": mysql_params["user"], + "ASPROM_TEST_DB_PASSWORD": mysql_params["passwd"], + "ASPROM_TEST_DB_NAME": mysql_params["db"], + } + ) + result = subprocess.run( + ["alembic", "upgrade", "head"], + cwd=ROOT, + env=env, + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 0, result.stderr diff --git a/tests/test_model_extras.py b/tests/test_model_extras.py new file mode 100644 index 0000000..d2dbba7 --- /dev/null +++ b/tests/test_model_extras.py @@ -0,0 +1,22 @@ +"""Tests for AspromModel HTML changelog rendering.""" + +from inc.asprom import AspromScheduleModel + + +def test_get_last_log_html(seed_service, asprom_model): + svc = seed_service(port=5500, product="http") + svc.approve("test justification", "auditor") + html = asprom_model.getLastLog(5) + assert "5500" in html + assert "test justification" in html + assert "auditor" in html + + +def test_get_scanned_ranges_plaintext(): + instance = AspromScheduleModel.__new__(AspromScheduleModel) + instance.schedule = [ + {"iprange": "10.0.0.0/24"}, + {"iprange": "192.168.0.0/16"}, + ] + instance.getSchedule = lambda: instance.schedule + assert instance.getScannedRanges() == "10.0.0.0/24\n192.168.0.0/16" diff --git a/tests/test_nagios_check.py b/tests/test_nagios_check.py new file mode 100644 index 0000000..3e6b62e --- /dev/null +++ b/tests/test_nagios_check.py @@ -0,0 +1,48 @@ +"""Tests for aspromNagiosCheck exit codes.""" + +from unittest.mock import patch + +import pytest + +from inc.asprom import genMessages + + +def test_nagios_check_ok(): + with patch("aspromNagiosCheck.AspromModel") as mock_model: + mock_model.return_value.getAlertsExposed.return_value = [] + mock_model.return_value.getAlertsClosed.return_value = [] + with patch("aspromNagiosCheck.genMessages", return_value=([], [])): + with patch("aspromNagiosCheck.initDB"): + with patch("aspromNagiosCheck.closeDB"): + with patch("aspromNagiosCheck.Cfg") as mock_cfg: + mock_cfg.return_value.__getitem__ = lambda s, k: { + "misc": {"url": "http://x"} + }[k] + with patch("aspromNagiosCheck.logger"): + with pytest.raises(SystemExit) as exc: + import aspromNagiosCheck + + aspromNagiosCheck.main() + assert exc.value.code == 0 + + +def test_nagios_exit_logic_critical(): + rows = [{"service": "ssh", "port": 22, "hostname": "h", "ip": "1.1.1.1", "crit": True}] + crit, warn = genMessages(rows) + exitstate = 0 + if len(crit): + exitstate = 2 + elif len(warn): + exitstate = 1 + assert exitstate == 2 + + +def test_nagios_exit_logic_warning(): + rows = [{"service": "ssh", "port": 22, "hostname": "h", "ip": "1.1.1.1", "crit": False}] + crit, warn = genMessages(rows) + exitstate = 0 + if len(crit): + exitstate = 2 + elif len(warn): + exitstate = 1 + assert exitstate == 1 diff --git a/tests/test_scan.py b/tests/test_scan.py new file mode 100644 index 0000000..cf4e6ec --- /dev/null +++ b/tests/test_scan.py @@ -0,0 +1,59 @@ +"""Integration tests for scan().""" + +from unittest.mock import MagicMock, patch + +from inc.asprom import scan + + +def _mock_nmap_result(): + mock_ps = MagicMock() + mock_ps.all_hosts.return_value = ["10.0.0.5"] + mock_ps.__getitem__ = lambda self, host: { + "hostname": "scanned-host", + "tcp": { + 22: {"state": "open", "product": "ssh", "version": "2.0", "extrainfo": ""}, + 80: {"state": "closed", "product": "", "version": "", "extrainfo": ""}, + }, + } + return mock_ps + + +@patch("inc.asprom.nmap.PortScanner") +def test_scan_creates_machine_and_service(mock_scanner, db_connection): + mock_scanner.return_value = _mock_nmap_result() + + state = scan("10.0.0.5", "22", "", "test-job-id") + assert state == "OK" + + cur = db_connection.cursor() + cur.execute("SELECT COUNT(*) FROM machines WHERE ip = %s", ("10.0.0.5",)) + assert cur.fetchone()[0] == 1 + + cur.execute( + """SELECT COUNT(*) FROM services s + INNER JOIN machines m ON s.machineId = m.id + WHERE m.ip = %s AND s.port = 22""", + ("10.0.0.5",), + ) + assert cur.fetchone()[0] == 1 + + cur.execute( + "SELECT state FROM scanlog WHERE jobid = %s ORDER BY id DESC LIMIT 1", + ("test-job-id",), + ) + assert cur.fetchone()[0] == "OK" + + +@patch("inc.asprom.nmap.PortScanner") +def test_scan_marks_stale_in_progress_as_timeout(mock_scanner, db_connection): + mock_scanner.return_value = _mock_nmap_result() + cur = db_connection.cursor() + cur.execute( + """INSERT INTO scanlog (jobid, state, startdate, iprange) + VALUES ('stale', 'IN PROGRESS', DATE_SUB(NOW(), INTERVAL 2 DAY), '10.0.0.0/24')""" + ) + db_connection.commit() + + scan("10.0.0.5", "22", "", "fresh-job") + cur.execute("SELECT state FROM scanlog WHERE jobid = 'stale'") + assert cur.fetchone()[0] == "TIMEOUT" diff --git a/tests/test_schedule.py b/tests/test_schedule.py new file mode 100644 index 0000000..b04a61b --- /dev/null +++ b/tests/test_schedule.py @@ -0,0 +1,66 @@ +"""Tests for schedule model helpers.""" + +from datetime import datetime +from unittest.mock import MagicMock + +import pytest + +from inc.asprom import AspromScheduleModel, NoJoibIDException + + +def test_promote_to_index_dicts(): + rows = [ + {"id": "a", "ip": "10.0.0.1", "port": "22"}, + {"id": "b", "ip": "10.0.0.2", "port": "80"}, + ] + result = AspromScheduleModel.promoteToIndex(rows, "id") + assert result["a"]["ip"] == "10.0.0.1" + assert result["b"]["port"] == "80" + + +def test_promote_to_index_lists(): + rows = [[1, 2, 3], [4, 5, 6]] + result = AspromScheduleModel.promoteToIndex(rows, 1) + assert result[2] == [1, 3] + assert result[5] == [4, 6] + + +def test_fetch_job_parses_command(): + sm = AspromScheduleModel.__new__(AspromScheduleModel) + sm.scheduleLog = {} + + job = MagicMock() + job.is_enabled.return_value = True + job.command = ( + "python /asprom/aspromScan.py -j " + '550e8400-e29b-41d4-a716-446655440000 -o="-sV" -p 1-1024 10.0.0.0/24' + ) + job.slices.render.return_value = "0 0 * * *" + job.schedule.return_value.get_next.return_value = datetime(2026, 1, 1, 12, 0) + + spec = sm._AspromScheduleModel__fetchJob(job) + assert spec["id"] == "550e8400-e29b-41d4-a716-446655440000" + assert spec["iprange"] == "10.0.0.0/24" + assert spec["ports"] == "1-1024" + assert spec["params"] == "-sV" + + +def test_fetch_job_raises_for_non_asprom(): + sm = AspromScheduleModel.__new__(AspromScheduleModel) + sm.scheduleLog = {} + job = MagicMock() + job.is_enabled.return_value = False + job.command = "/bin/true" + with pytest.raises(NoJoibIDException): + sm._AspromScheduleModel__fetchJob(job) + + +def test_get_scanned_ranges(): + sm = AspromScheduleModel.__new__(AspromScheduleModel) + sm.schedule = [ + {"iprange": "10.0.0.0/24"}, + {"iprange": "192.168.1.0/24"}, + ] + sm.getSchedule = lambda: sm.schedule + result = sm.getScannedRanges() + assert result == "10.0.0.0/24\n192.168.1.0/24" diff --git a/tests/test_schedule_jobs.py b/tests/test_schedule_jobs.py new file mode 100644 index 0000000..74252a9 --- /dev/null +++ b/tests/test_schedule_jobs.py @@ -0,0 +1,45 @@ +"""Tests for AspromScheduleModel job management with mocked crontab.""" + +from unittest.mock import MagicMock, patch + +from inc.asprom import AspromScheduleModel + + +@patch.object(AspromScheduleModel, "render") +@patch.object(AspromScheduleModel, "write") +@patch.object(AspromScheduleModel, "read") +def test_change_job_updates_command(mock_read, mock_write, mock_render): + sm = AspromScheduleModel.__new__(AspromScheduleModel) + sm.jobsByID = {} + job = MagicMock() + job.is_enabled.return_value = True + sm.getJobByID = MagicMock(return_value=job) + + with patch("inc.asprom.get_cfg") as mock_cfg: + mock_cfg.return_value.maindir = "/asprom" + sm.changeJob( + "550e8400-e29b-41d4-a716-446655440000", + "0 0 * * *", + "10.0.0.0/24", + "22", + "-sV", + job=job, + ) + + job.set_command.assert_called_once() + assert "10.0.0.0/24" in job.set_command.call_args[0][0] + mock_write.assert_called_once() + mock_read.assert_called_once() + + +@patch.object(AspromScheduleModel, "write") +@patch.object(AspromScheduleModel, "read") +def test_delete_job_disables_entry(mock_read, mock_write): + sm = AspromScheduleModel.__new__(AspromScheduleModel) + job = MagicMock() + sm.getJobByID = MagicMock(return_value=job) + + sm.deleteJob("550e8400-e29b-41d4-a716-446655440000") + + job.enable.assert_called_once_with(False) + mock_write.assert_called_once() diff --git a/tests/test_service_lifecycle.py b/tests/test_service_lifecycle.py new file mode 100644 index 0000000..75f1a12 --- /dev/null +++ b/tests/test_service_lifecycle.py @@ -0,0 +1,25 @@ +"""Integration tests for Service lifecycle.""" + +from tests.helpers import count_servicelog_entries, current_service_open + + +def test_service_create_opens_servicelog(seed_service, db_connection): + svc = seed_service(port=22) + assert svc.port == 22 + assert current_service_open(db_connection, svc.id) is True + + +def test_service_create_and_delete_lifecycle(seed_service, db_connection): + svc = seed_service(port=8080, product="http-proxy") + assert current_service_open(db_connection, svc.id) is True + assert count_servicelog_entries(db_connection, svc.id) >= 1 + + svc.delete() + assert current_service_open(db_connection, svc.id) is False + + +def test_service_create_idempotent(seed_service, db_connection): + mach = seed_service(port=22).machine + svc1 = seed_service(port=22, machine=mach) + svc2 = seed_service(port=22, machine=mach) + assert svc1.id == svc2.id diff --git a/tests/test_views.py b/tests/test_views.py new file mode 100644 index 0000000..0517d0d --- /dev/null +++ b/tests/test_views.py @@ -0,0 +1,36 @@ +"""Integration tests for SQL views.""" + + +def test_exposed_view_lists_open_unapproved(seed_service, db_connection, asprom_model): + exposed = seed_service(port=10001, product="test-a") + approved = seed_service(port=10002, product="test-b") + approved.approve("ok", "user") + + rows = asprom_model.getAlertsExposed() + ids = [r["id"] for r in rows] + assert exposed.id in ids + assert approved.id not in ids + + +def test_servicelog_cur_reflects_latest_state(seed_service, db_connection): + svc = seed_service(port=10003) + cur = db_connection.cursor() + cur.execute("SELECT openp FROM servicelogCur WHERE serviceId = %s", (svc.id,)) + assert cur.fetchone()[0] == 1 + + svc.delete() + cur.execute("SELECT openp FROM servicelogCur WHERE serviceId = %s", (svc.id,)) + assert cur.fetchone()[0] == 0 + + +def test_neatline_view_after_approval(seed_service, db_connection): + svc = seed_service(port=10004) + svc.approve("justified", "user") + cur = db_connection.cursor() + cur.execute( + "SELECT neat, justification FROM neatline WHERE serviceId = %s", + (svc.id,), + ) + row = cur.fetchone() + assert row[0] == 1 + assert row[1] == "justified"