Skip to content

feat: richm fprint #233

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
richm wants to merge 3 commits into from
Closed

feat: richm fprint #233

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5cb6417
feat: Write roles fingerprints to /var/log/sysroles.jsonl
spetrosi Aug 5, 2026
be79445
fix: Update sr_fingerprint task calls to use new structured parameters
spetrosi Aug 5, 2026
5e3421a
feat: suggested fixes for sr_fingerprint log file
richm Aug 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions defaults/main.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,3 +9,5 @@ bootloader_reboot_ok: false

bootloader_gather_facts: false
bootloader_secure_logging: true

bootloader_write_log_file: false
Comment thread
coderabbitai[bot] marked this conversation as resolved.
322 changes: 298 additions & 24 deletions library/sr_fingerprint.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,29 +7,148 @@
DOCUMENTATION = """
---
module: sr_fingerprint
short_description: Write a message string to syslog using Ansible C(module.log) function.
short_description: Write role fingerprint data to syslog and optionally to a JSONL log file
description:
- Writes the given string to the system log using Ansible C(module.log) function.
- Collects role fingerprint data into a canonical record and writes it to
syslog using Ansible C(module.log) as C(key=value) pairs.
- Optionally appends the same record as a JSON line to a log file
(one JSON object per line, JSONL format), by default
C(/var/log/sysroles.jsonl).
- Playbook variables are not available inside modules automatically. Roles
pass C(role_name), C(role_path), C(ansible_play_hosts_all),
C(distribution), and C(distribution_version) from the task.
- C(ansible_check_mode) is collected from the module execution context.
- Intended for role-internal or diagnostic use.
author: Rich Megginson (@richm)
options:
sr_message:
description: Text to record in syslog.
status:
description: Role execution status.
type: str
required: true
choices:
- begin
- success
write_log_file:
description: >-
If C(true), append fingerprint data to the JSONL log file.
Defaults to C(false).
type: bool
default: false
log_file:
description: >-
Path to the JSONL log file. A lock sidecar (C(<log_file>.lock))
is created next to the log file for cross-process safety.
type: path
default: /var/log/sysroles.jsonl
max_log_size:
description: >-
Maximum log file size in bytes. When appending a new record
would exceed this limit, the oldest records are removed first.
Set to C(0) to disable trimming.
type: int
default: 2000000
role_name:
description: Name of the role, typically C({{ role_name }}).
type: str
required: true
role_path:
description: Path to the role, typically C({{ role_path }}).
type: path
required: true
ansible_play_hosts_all:
description: >-
All hosts in the play, typically C({{ ansible_play_hosts_all }}).
Used to derive C(play_hosts_number).
type: list
elements: str
required: true
distribution:
description: >-
OS distribution name, typically
C({{ ansible_facts["distribution"] }}).
type: str
default: ""
distribution_version:
description: >-
OS distribution version, typically
C({{ ansible_facts["distribution_version"] }}).
type: str
default: ""
"""

EXAMPLES = """
- name: Record a fingerprint message in syslog
- name: Record role begin fingerprint to syslog only (not log file)
sr_fingerprint:
status: begin
role_name: bootloader
role_path: "{{ role_path }}"
ansible_play_hosts_all: "{{ ansible_play_hosts_all }}"
distribution: "{{ ansible_facts['distribution'] }}"
distribution_version: "{{ ansible_facts['distribution_version'] }}"
write_log_file: false

- name: Record role success fingerprint
sr_fingerprint:
sr_message: "system_role:ROLENAME"
status: success
role_name: bootloader
role_path: "{{ role_path }}"
ansible_play_hosts_all: "{{ ansible_play_hosts_all }}"
distribution: "{{ ansible_facts['distribution'] }}"
distribution_version: "{{ ansible_facts['distribution_version'] }}"
write_log_file: true
"""

RETURN = r""" # """
RETURN = r"""
fingerprint:
description: The fingerprint record written to syslog and optionally to the log file.
returned: always
type: dict
sample:
date: "2026-08-03T10:15:00+02:00"
role_name: network
role_path: /usr/share/ansible/roles/linux-system-roles.network
status: success
ansible_version: "2.16.3"
managed_node_distro: RedHat-9.4
play_hosts_number: 3
ansible_check_mode: false
message:
description: Informational message shown in check mode.
returned: check mode
type: str
sample: "Check mode: message not logged - [date=... role_name=...]"
jsonl_row:
description: The JSON line that would be appended to the log file.
returned: check mode and O(write_log_file=true)
type: str
log_file:
description: Path to the log file that would be written.
returned: check mode and O(write_log_file=true)
type: str
"""

from ansible.module_utils.basic import AnsibleModule

import datetime
import errno
import fcntl
import json
import os
import stat
import tempfile

FINGERPRINT_FIELDS = (
"date",
"role_name",
"role_path",
"status",
"ansible_version",
"managed_node_distro",
"play_hosts_number",
"ansible_check_mode",
)

FINGERPRINT_SYSLOG_SEPARATOR = " "


def _local_iso8601_no_microseconds():
Expand All @@ -51,33 +170,188 @@
return datetime.datetime.now(utc).astimezone().replace(microsecond=0).isoformat()


def run_module():
module_args = dict(
sr_message=dict(type="str", required=True),
)
def _ensure_parent_dir(path):
parent = os.path.dirname(path)
if not parent:
return
if os.path.isdir(parent):
return
try:
os.makedirs(parent)
except OSError as exc:
if exc.errno != errno.EEXIST or not os.path.isdir(parent):
raise

module = AnsibleModule(
argument_spec=module_args,
supports_check_mode=True,
)

log_message = "%s %s" % (
module.params["sr_message"],
_local_iso8601_no_microseconds(),
)
def _format_fingerprint_jsonl(record):
"""Format the canonical fingerprint record as a single JSON line."""
return json.dumps(record, separators=(",", ":"), sort_keys=False)


def _trim_log_file(log_file, size_needed):
"""Remove oldest records until the file can accommodate size_needed bytes."""
with open(log_file, "r") as log_fd:
lines = log_fd.readlines()
size_removed = 0
while lines and size_removed < size_needed:
size_removed += len(lines.pop(0))
orig_stat = os.stat(log_file)
dir_name = os.path.dirname(log_file) or "."
fd, tmp_path = tempfile.mkstemp(dir=dir_name, suffix=".tmp")
try:
os.fchmod(fd, stat.S_IMODE(orig_stat.st_mode))
try:
os.fchown(fd, orig_stat.st_uid, orig_stat.st_gid)
except OSError:

Check notice

Code scanning / CodeQL

Empty except Note

'except' clause does nothing but pass and there is no explanatory comment.
pass
with os.fdopen(fd, "w") as tmp_fd:
tmp_fd.writelines(lines)
tmp_fd.flush()
os.fsync(tmp_fd.fileno())
os.rename(tmp_path, log_file)
except BaseException:
try:
os.unlink(tmp_path)
except OSError:

Check notice

Code scanning / CodeQL

Empty except Note

'except' clause does nothing but pass and there is no explanatory comment.
pass
raise


def _write_jsonl_log(log_file, record, max_size=0):
_ensure_parent_dir(log_file)
new_line = _format_fingerprint_jsonl(record) + "\n"
lock_path = log_file + ".lock"
lock_fd = open(lock_path, "w")
try:
fcntl.flock(lock_fd, fcntl.LOCK_EX)
try:
cur_size = os.path.getsize(log_file)
except OSError:
cur_size = 0
if max_size > 0 and cur_size + len(new_line) > max_size and cur_size > 0:
_trim_log_file(log_file, len(new_line))
Comment on lines +191 to +232

@coderabbitai coderabbitai Bot Aug 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

Trim does not enforce max_log_size for an already oversized file.

_write_jsonl_log calls _trim_log_file(log_file, len(new_line)), and _trim_log_file stops after it removes len(new_line) bytes. The final size is then cur_size - removed + len(new_line), which stays at about cur_size. If cur_size already exceeds max_size, the file never returns to the limit.

Two paths reach that state:

  • A single record larger than max_size is written while the file is empty, because line 231 requires cur_size > 0.
  • The file was written earlier with a larger max_log_size or with trimming disabled.

Pass the required reduction instead of the new line size.

🔧 Proposed fix to trim down to the limit 
-        if max_size > 0 and cur_size + len(new_line) > max_size and cur_size > 0:
-            _trim_log_file(log_file, len(new_line))
+        if max_size > 0 and cur_size + len(new_line) > max_size and cur_size > 0:
+            _trim_log_file(log_file, cur_size + len(new_line) - max_size)
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
def _trim_log_file(log_file, size_needed):
"""Remove oldest records until the file can accommodate size_needed bytes."""
with open(log_file, "r") as log_fd:
lines = log_fd.readlines()
size_removed = 0
while lines and size_removed < size_needed:
size_removed += len(lines.pop(0))
orig_stat = os.stat(log_file)
dir_name = os.path.dirname(log_file) or "."
fd, tmp_path = tempfile.mkstemp(dir=dir_name, suffix=".tmp")
try:
os.fchmod(fd, stat.S_IMODE(orig_stat.st_mode))
try:
os.fchown(fd, orig_stat.st_uid, orig_stat.st_gid)
except OSError:
pass
with os.fdopen(fd, "w") as tmp_fd:
tmp_fd.writelines(lines)
tmp_fd.flush()
os.fsync(tmp_fd.fileno())
os.rename(tmp_path, log_file)
except BaseException:
try:
os.unlink(tmp_path)
except OSError:
pass
raise
def _write_jsonl_log(log_file, record, max_size=0):
_ensure_parent_dir(log_file)
new_line = _format_fingerprint_jsonl(record) + "\n"
lock_path = log_file + ".lock"
lock_fd = open(lock_path, "w")
try:
fcntl.flock(lock_fd, fcntl.LOCK_EX)
try:
cur_size = os.path.getsize(log_file)
except OSError:
cur_size = 0
if max_size > 0 and cur_size + len(new_line) > max_size and cur_size > 0:
_trim_log_file(log_file, len(new_line))
def _trim_log_file(log_file, size_needed):
"""Remove oldest records until the file can accommodate size_needed bytes."""
with open(log_file, "r") as log_fd:
lines = log_fd.readlines()
size_removed = 0
while lines and size_removed < size_needed:
size_removed += len(lines.pop(0))
orig_stat = os.stat(log_file)
dir_name = os.path.dirname(log_file) or "."
fd, tmp_path = tempfile.mkstemp(dir=dir_name, suffix=".tmp")
try:
os.fchmod(fd, stat.S_IMODE(orig_stat.st_mode))
try:
os.fchown(fd, orig_stat.st_uid, orig_stat.st_gid)
except OSError:
pass
with os.fdopen(fd, "w") as tmp_fd:
tmp_fd.writelines(lines)
tmp_fd.flush()
os.fsync(tmp_fd.fileno())
os.rename(tmp_path, log_file)
except BaseException:
try:
os.unlink(tmp_path)
except OSError:
pass
raise
def _write_jsonl_log(log_file, record, max_size=0):
_ensure_parent_dir(log_file)
new_line = _format_fingerprint_jsonl(record) + "\n"
lock_path = log_file + ".lock"
lock_fd = open(lock_path, "w")
try:
fcntl.flock(lock_fd, fcntl.LOCK_EX)
try:
cur_size = os.path.getsize(log_file)
except OSError:
cur_size = 0
if max_size > 0 and cur_size + len(new_line) > max_size and cur_size > 0:
_trim_log_file(log_file, cur_size + len(new_line) - max_size)
🧰 Tools
🪛 ast-grep (0.45.0)

[warning] 192-192: File path is request-/variable-derived; validate and normalize to prevent path traversal.
Context: open(log_file, "r")
Note: [CWE-22] Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal').

(open-filename-from-request)

[warning] 223-223: File path is request-/variable-derived; validate and normalize to prevent path traversal.
Context: open(lock_path, "w")
Note: [CWE-22] Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal').

(open-filename-from-request)

[warning] 232-232: File path is request-/variable-derived; validate and normalize to prevent path traversal.
Context: open(log_file, "a")
Note: [CWE-22] Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal').

(open-filename-from-request)

🪛 GitHub Check: CodeQL

[notice] 205-205: Empty except
'except' clause does nothing but pass and there is no explanatory comment.

[notice] 215-215: Empty except
'except' clause does nothing but pass and there is no explanatory comment.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@library/sr_fingerprint.py` around lines 191 - 232, Update _write_jsonl_log to
calculate the required reduction as cur_size + len(new_line) - max_size, and
pass that value to _trim_log_file instead of len(new_line). Apply trimming
whenever max_size is exceeded, including an empty file with an oversized record,
while preserving normal writes when the limit is not exceeded.

with open(log_file, "a") as log_fd:
log_fd.write(new_line)
finally:
fcntl.flock(lock_fd, fcntl.LOCK_UN)
lock_fd.close()


def _get_managed_node_distro(distribution, distribution_version):
if distribution and distribution_version:
return "%s-%s" % (distribution, distribution_version)
return "unknown"


def _get_play_hosts_number(play_hosts_all):
return len(play_hosts_all)


def _get_ansible_version(module):
version = getattr(module, "ansible_version", None)
if version:
return version
return "unknown"


def _get_check_mode(module):
return bool(getattr(module, "check_mode", False))


def _collect_fingerprint_record(module, status):
"""Build the canonical fingerprint record used by all output formatters."""
return {
"date": _local_iso8601_no_microseconds(),
"role_name": module.params["role_name"],
"role_path": module.params["role_path"],
"status": status,
"ansible_version": _get_ansible_version(module),
"managed_node_distro": _get_managed_node_distro(
module.params["distribution"], module.params["distribution_version"]
),
"play_hosts_number": _get_play_hosts_number(
module.params["ansible_play_hosts_all"]
),
"ansible_check_mode": _get_check_mode(module),
}


def _fingerprint_record_items(record):
return [(field, record[field]) for field in FINGERPRINT_FIELDS]


def _format_fingerprint_key_value(field, value):
text = "" if value is None else str(value)
if any(char in text for char in ' "='):
return '%s="%s"' % (field, text.replace('"', '""'))
return "%s=%s" % (field, text)


def _format_fingerprint_syslog(record):
"""Format the canonical fingerprint record as key=value syslog text."""
pairs = [
_format_fingerprint_key_value(field, value)
for field, value in _fingerprint_record_items(record)
]
return FINGERPRINT_SYSLOG_SEPARATOR.join(pairs)


def _handle_fingerprint(module):
max_log_size = module.params["max_log_size"]
if max_log_size < 0:
module.fail_json(
msg="max_log_size must be 0 or a positive integer, got %d" % max_log_size
)

fingerprint_record = _collect_fingerprint_record(module, module.params["status"])
log_message = _format_fingerprint_syslog(fingerprint_record)

if module.check_mode:
module.exit_json(
result = dict(
changed=False,
message="Check mode: message not logged - [%s]" % log_message,
fingerprint=fingerprint_record,
)
if module.params["write_log_file"]:
result["jsonl_row"] = _format_fingerprint_jsonl(fingerprint_record)
result["log_file"] = module.params["log_file"]
module.exit_json(**result)

module.log(log_message)

# we don't actually change anything, so we're not changed - writing a log message
# is not considered a change
# also, we don't want to report changed every time the role runs
module.exit_json(changed=False)
if module.params["write_log_file"]:
log_file = module.params["log_file"]
try:
_write_jsonl_log(
log_file, fingerprint_record, module.params["max_log_size"]
)
except (IOError, OSError) as exc:
module.fail_json(
msg="Failed to write fingerprint log file %s: %s" % (log_file, exc)
)

module.exit_json(changed=False, fingerprint=fingerprint_record)


def run_module():
module_args = dict(
status=dict(type="str", required=True, choices=["begin", "success"]),
write_log_file=dict(type="bool", default=False),
log_file=dict(type="path", default="/var/log/sysroles.jsonl"),
max_log_size=dict(type="int", default=2000000),
role_name=dict(type="str", required=True),
role_path=dict(type="path", required=True),
ansible_play_hosts_all=dict(type="list", elements="str", required=True),
distribution=dict(type="str", default=""),
distribution_version=dict(type="str", default=""),
)

module = AnsibleModule(
argument_spec=module_args,
supports_check_mode=True,
)

_handle_fingerprint(module)


def main():
Expand Down
10 changes: 7 additions & 3 deletions tasks/main.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,10 @@

- name: Record role success fingerprint
sr_fingerprint:
sr_message: >-
success system_role:bootloader ansible_version={{ ansible_version.full }}
{{ ansible_facts['distribution'] }}-{{ ansible_facts['distribution_version'] }}
status: success
role_name: bootloader
role_path: "{{ role_path }}"
ansible_play_hosts_all: "{{ ansible_play_hosts_all }}"
distribution: "{{ ansible_facts['distribution'] }}"
distribution_version: "{{ ansible_facts['distribution_version'] }}"
write_log_file: "{{ bootloader_write_log_file }}"
10 changes: 7 additions & 3 deletions tasks/set_vars.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,9 +8,13 @@

- name: Record role begin fingerprint
sr_fingerprint:
sr_message: >-
begin system_role:bootloader ansible_version={{ ansible_version.full }}
{{ ansible_facts['distribution'] }}-{{ ansible_facts['distribution_version'] }}
status: begin
role_name: bootloader
role_path: "{{ role_path }}"
ansible_play_hosts_all: "{{ ansible_play_hosts_all }}"
distribution: "{{ ansible_facts['distribution'] }}"
distribution_version: "{{ ansible_facts['distribution_version'] }}"
Comment on lines +15 to +16

@coderabbitai coderabbitai Bot Aug 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win

Unguarded distribution fact lookups in both fingerprint tasks. Both sr_fingerprint tasks read ansible_facts['distribution'] and ansible_facts['distribution_version'] without a default. The setup task in tasks/set_vars.yml runs only when required facts are missing, so a pre-gathered fact subset that excludes distribution makes these lookups undefined and fails the task. The module already defaults both parameters to "".

  • tasks/set_vars.yml#L15-L16: add | d('') to both fact lookups in the begin fingerprint task.
  • tasks/main.yml#L186-L187: add | d('') to both fact lookups in the success fingerprint task.
📍 Affects 2 files
  • tasks/set_vars.yml#L15-L16 (this comment)
  • tasks/main.yml#L186-L187
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tasks/set_vars.yml` around lines 15 - 16, Guard the distribution fact lookups
in both fingerprint tasks with an empty-string default: update
tasks/set_vars.yml lines 15-16 and tasks/main.yml lines 186-187 for the begin
and success fingerprint tasks, respectively. Apply the default to both
distribution and distribution_version while preserving the existing module
parameter flow.

write_log_file: "{{ bootloader_write_log_file }}"

- name: Determine if system is ostree and set flag
when: not __bootloader_is_ostree is defined
Expand Down
Loading
Loading