377 lines
11 KiB
Python
377 lines
11 KiB
Python
"""A git-based audit backend
|
|
|
|
The backend minimizes commits by caching changes until an already
|
|
changed record is changed again. In this case all changes are
|
|
committed.
|
|
|
|
Changes are annotated with a time stamp and a user-id
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import time
|
|
from datetime import (
|
|
UTC,
|
|
datetime,
|
|
)
|
|
from pathlib import Path
|
|
from threading import (
|
|
Lock,
|
|
Thread,
|
|
)
|
|
|
|
import yaml
|
|
from datalad_core.git_utils import apply_changeset
|
|
from datalad_core.repo import Repo
|
|
from datalad_core.runners import (
|
|
CommandError,
|
|
call_git,
|
|
)
|
|
|
|
from dump_things_service.audit import AuditBackend
|
|
|
|
index_file_name = 'gitaudit_index.log'
|
|
|
|
GIT_ERROR_UNCLEAN_EXIT = 128
|
|
|
|
|
|
class FlushingThread(Thread):
|
|
def __init__(
|
|
self,
|
|
backend: GitAuditBackend,
|
|
auto_flush_timeout: int,
|
|
):
|
|
super().__init__()
|
|
self.auto_flush_timeout = auto_flush_timeout
|
|
self.backend = backend
|
|
self.exit_requested = False
|
|
self.daemon = True
|
|
|
|
def request_exit(self):
|
|
self.exit_requested = True
|
|
|
|
def run(self):
|
|
while not self.exit_requested:
|
|
time.sleep(1)
|
|
if time.time() - self.backend.last_flush_time > self.auto_flush_timeout:
|
|
self.backend.flush()
|
|
|
|
|
|
class GitAuditBackend(AuditBackend):
|
|
def __init__(
|
|
self,
|
|
path: Path,
|
|
auto_flush_timeout: int = 60,
|
|
):
|
|
self.path = path
|
|
self.index_path: Path | None = None
|
|
self.cached_index_entries: list[str] = []
|
|
self.current_change_set: dict[Path, str] = {}
|
|
self.lock = Lock()
|
|
self.last_flush_time = 0
|
|
if auto_flush_timeout < 1:
|
|
msg = 'auto_flush_timeout must be greater or equal to 1'
|
|
raise ValueError(msg)
|
|
self.flushing_thread = FlushingThread(self, auto_flush_timeout)
|
|
self.flushing_thread.start()
|
|
self._init_repo()
|
|
|
|
def __del__(self):
|
|
self.flush()
|
|
if self.flushing_thread:
|
|
self.flushing_thread.request_exit()
|
|
self.flushing_thread.join()
|
|
self.flushing_thread = None
|
|
|
|
def add_record(
|
|
self,
|
|
record: dict,
|
|
committer_id: str,
|
|
author_id: str | None = None,
|
|
) -> None:
|
|
with self.lock:
|
|
author_id = committer_id if author_id is None else author_id
|
|
record_id = record['pid']
|
|
location = self._get_location_for(record_id)
|
|
if self._has_pending_changes(location):
|
|
self._persist_pending_changes()
|
|
self._add_elements(record_id, location, committer_id, author_id, record)
|
|
|
|
def flush(self):
|
|
with self.lock:
|
|
self._locked_flush()
|
|
|
|
def _locked_flush(self):
|
|
if self.current_change_set:
|
|
self._persist_pending_changes()
|
|
if self.cached_index_entries:
|
|
with self.index_path.open('at') as f:
|
|
f.write('\n'.join(self.cached_index_entries) + '\n')
|
|
self.cached_index_entries = []
|
|
self.last_flush_time = time.time()
|
|
|
|
def get_audit_log(
|
|
self,
|
|
record_id: str,
|
|
) -> dict:
|
|
with self.lock:
|
|
return self._locked_get_audit_log(record_id)
|
|
|
|
def _locked_get_audit_log(
|
|
self,
|
|
record_id: str,
|
|
) -> dict:
|
|
self._locked_flush()
|
|
|
|
# Get all commits that updated the log. Those will also have updated
|
|
# the records
|
|
changes = []
|
|
yaml_location, log_location = map(str, self._get_location_for(record_id)[1:])
|
|
commit_hashes = (
|
|
call_git(
|
|
['log', '--format=%H', '--', log_location],
|
|
cwd=self.path,
|
|
capture_output=True,
|
|
)
|
|
.decode()
|
|
.splitlines()
|
|
)
|
|
for commit_hash in commit_hashes:
|
|
log_diff_lines = (
|
|
call_git(
|
|
['show', '--format=%b', commit_hash, '--', log_location],
|
|
cwd=self.path,
|
|
capture_output=True,
|
|
)
|
|
.decode()
|
|
.splitlines()
|
|
)
|
|
# Get the log entry
|
|
log_line = next(
|
|
filter(
|
|
lambda line: not line.startswith('+++') and line.startswith('+'),
|
|
log_diff_lines,
|
|
)
|
|
)[1:]
|
|
log_entry = json.loads(log_line)
|
|
|
|
# Get the YAML diff
|
|
yaml_diff_lines = (
|
|
call_git(
|
|
['show', '--format=%b', commit_hash, '--', yaml_location],
|
|
cwd=self.path,
|
|
capture_output=True,
|
|
)
|
|
.decode()
|
|
.splitlines()
|
|
)
|
|
yaml_diff = (
|
|
'\n'.join(filter(lambda line: line != '', yaml_diff_lines)) + '\n'
|
|
)
|
|
|
|
# Get the YAML content
|
|
yaml_content = call_git(
|
|
['show', f'{commit_hash}:{yaml_location}'],
|
|
cwd=self.path,
|
|
capture_output=True,
|
|
).decode()
|
|
changes.append(
|
|
(
|
|
log_entry['time_stamp'],
|
|
log_entry['committer_id'],
|
|
log_entry['author_id'],
|
|
yaml_diff,
|
|
yaml_content,
|
|
)
|
|
)
|
|
|
|
changes.sort()
|
|
return {c[0]: c[1:] for c in changes}
|
|
|
|
def get_audit_logs(
|
|
self,
|
|
record_id_pattern: str,
|
|
) -> dict:
|
|
with self.lock:
|
|
return self._locked_get_audit_logs(record_id_pattern)
|
|
|
|
def _locked_get_audit_logs(
|
|
self,
|
|
record_id_pattern: str,
|
|
) -> dict:
|
|
self._locked_flush()
|
|
matcher = re.compile(record_id_pattern)
|
|
matching_ids = tuple(
|
|
filter(
|
|
lambda record_id: matcher.fullmatch(record_id) is not None,
|
|
self.index,
|
|
)
|
|
)
|
|
return {
|
|
record_id: self._locked_get_audit_log(record_id)
|
|
for record_id in sorted(matching_ids)
|
|
}
|
|
|
|
def _add_elements(
|
|
self,
|
|
record_id: str,
|
|
location: tuple[str, Path, Path],
|
|
committer_id: str,
|
|
author_id: str,
|
|
record: dict,
|
|
) -> bool:
|
|
from dump_things_service.utils import json2yaml # noqa PLC0415 -- global import leads to circular imports
|
|
|
|
existing_record = self._read_record_from_repo_path(location[1])
|
|
if existing_record != record:
|
|
self.current_change_set[location[1]] = json2yaml(record)
|
|
self._add_log_entry(location[2], committer_id, author_id)
|
|
self._add_index_entry(record_id)
|
|
return True
|
|
return False
|
|
|
|
def _add_log_entry(
|
|
self,
|
|
log_location: Path,
|
|
committer_id: str,
|
|
author_id: str,
|
|
) -> None:
|
|
time_stamp = datetime.now(tz=UTC).isoformat()
|
|
entry = {
|
|
'time_stamp': time_stamp,
|
|
'committer_id': committer_id,
|
|
'author_id': author_id,
|
|
}
|
|
log_content = self._read_from_repo_path(log_location).decode()
|
|
log_content += json.dumps(entry, ensure_ascii=False, sort_keys=True) + '\n'
|
|
self.current_change_set[log_location] = log_content
|
|
|
|
def _add_index_entry(
|
|
self,
|
|
record_id: str,
|
|
):
|
|
if record_id not in self.index:
|
|
self.cached_index_entries.append(record_id)
|
|
self.index.add(record_id)
|
|
|
|
def _read_from_repo_path(
|
|
self,
|
|
path: Path,
|
|
) -> bytes:
|
|
try:
|
|
return call_git(
|
|
['cat-file', '-p', f'master:{path!s}'],
|
|
cwd=self.path,
|
|
capture_output=True,
|
|
)
|
|
except CommandError as e:
|
|
if e.returncode == GIT_ERROR_UNCLEAN_EXIT:
|
|
return b''
|
|
raise
|
|
|
|
def _read_record_from_repo_path(
|
|
self,
|
|
path: Path,
|
|
):
|
|
return yaml.safe_load(self._read_from_repo_path(path))
|
|
|
|
def _has_pending_changes(
|
|
self,
|
|
location: tuple[str, Path, Path],
|
|
) -> bool:
|
|
log_pending = location[1] in self.current_change_set
|
|
record_pending = location[2] in self.current_change_set
|
|
if log_pending != record_pending:
|
|
msg = (
|
|
f'change status mismatch: changed: '
|
|
f'{location[1]} ({log_pending}), '
|
|
f'{location[2]} ({record_pending})'
|
|
)
|
|
raise SystemError(msg)
|
|
return log_pending
|
|
|
|
def _persist_pending_changes(self) -> None:
|
|
apply_changeset(
|
|
self.repo,
|
|
self.current_change_set,
|
|
message='persist changes',
|
|
)
|
|
self.current_change_set = {}
|
|
|
|
def _get_location_for(
|
|
self,
|
|
record_id: str,
|
|
) -> tuple[str, Path, Path]:
|
|
base = hashlib.sha1(record_id.encode()).hexdigest() # noqa S324 -- hash is not used for security
|
|
dir_1, dir_2, _name = base[0:3], base[3:6], base[6:]
|
|
location_dir = Path(dir_1) / Path(dir_2)
|
|
return (
|
|
base,
|
|
location_dir / (base + '.yaml'),
|
|
location_dir / (base + '.log'),
|
|
)
|
|
|
|
def _init_repo(self) -> None:
|
|
if self.path.exists():
|
|
is_empty = len(tuple(Path(self.path).glob('**'))) == 1
|
|
else:
|
|
self.path.mkdir(parents=True)
|
|
is_empty = True
|
|
self.index_path = self.path / index_file_name
|
|
|
|
if is_empty:
|
|
call_git(['init', '--bare', str(self.path)], capture_output=True)
|
|
self.repo = Repo(self.path)
|
|
call_git(['config', 'user.name', 'dump-things-service'], cwd=self.path)
|
|
call_git(
|
|
['config', 'user.email', 'dump-things-service@localhost'], cwd=self.path
|
|
)
|
|
apply_changeset(
|
|
self.repo,
|
|
{'README.txt': 'A git-based audit backend\n'},
|
|
message='add README.txt',
|
|
)
|
|
self.index_path.write_text('')
|
|
else:
|
|
self.repo = Repo(self.path)
|
|
|
|
if not self.index_path.exists():
|
|
self.rebuild_index()
|
|
|
|
with open(self.index_path) as f:
|
|
self.index = {line.strip() for line in f}
|
|
|
|
def _add_to_index(
|
|
self,
|
|
record_id: str,
|
|
):
|
|
if record_id not in self.index:
|
|
self.cached_index_entries.append(record_id)
|
|
self.index.add(record_id)
|
|
|
|
def rebuild_index(self):
|
|
tree_entries = (
|
|
call_git(
|
|
['ls-tree', '-r', 'master:'],
|
|
cwd=self.path,
|
|
capture_output=True,
|
|
)
|
|
.decode()
|
|
.splitlines()
|
|
)
|
|
with open(self.index_path, 'w') as f:
|
|
for line in tree_entries:
|
|
if not line.endswith('.yaml'):
|
|
continue
|
|
_flag, _object_type, object_hash, _file_name = line.split(maxsplit=3)
|
|
record = yaml.safe_load(
|
|
call_git(
|
|
['show', object_hash],
|
|
cwd=self.path,
|
|
capture_output=True,
|
|
).decode()
|
|
)
|
|
f.write(record['pid'] + '\n')
|