mirror of https://github.com/python/cpython
265 lines
8.3 KiB
Python
265 lines
8.3 KiB
Python
"""Support for documenting audit events."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import TYPE_CHECKING
|
|
|
|
from docutils import nodes
|
|
from sphinx.errors import NoUri
|
|
from sphinx.locale import _ as sphinx_gettext
|
|
from sphinx.transforms.post_transforms import SphinxPostTransform
|
|
from sphinx.util import logging
|
|
from sphinx.util.docutils import SphinxDirective
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Iterator
|
|
|
|
from sphinx.application import Sphinx
|
|
from sphinx.builders import Builder
|
|
from sphinx.environment import BuildEnvironment
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# This list of sets are allowable synonyms for event argument names.
|
|
# If two names are in the same set, they are treated as equal for the
|
|
# purposes of warning. This won't help if the number of arguments is
|
|
# different!
|
|
_SYNONYMS = [
|
|
frozenset({"file", "path", "fd"}),
|
|
]
|
|
|
|
|
|
class AuditEvents:
|
|
def __init__(self) -> None:
|
|
self.events: dict[str, list[str]] = {}
|
|
self.sources: dict[str, list[tuple[str, str]]] = {}
|
|
|
|
def __iter__(self) -> Iterator[tuple[str, list[str], tuple[str, str]]]:
|
|
for name, args in self.events.items():
|
|
for source in self.sources[name]:
|
|
yield name, args, source
|
|
|
|
def add_event(
|
|
self, name, args: list[str], source: tuple[str, str]
|
|
) -> None:
|
|
if name in self.events:
|
|
self._check_args_match(name, args)
|
|
else:
|
|
self.events[name] = args
|
|
self.sources.setdefault(name, []).append(source)
|
|
|
|
def _check_args_match(self, name: str, args: list[str]) -> None:
|
|
current_args = self.events[name]
|
|
msg = (
|
|
f"Mismatched arguments for audit-event {name}: "
|
|
f"{current_args!r} != {args!r}"
|
|
)
|
|
if current_args == args:
|
|
return
|
|
if len(current_args) != len(args):
|
|
logger.warning(msg)
|
|
return
|
|
for a1, a2 in zip(current_args, args, strict=False):
|
|
if a1 == a2:
|
|
continue
|
|
if any(a1 in s and a2 in s for s in _SYNONYMS):
|
|
continue
|
|
logger.warning(msg)
|
|
return
|
|
|
|
def id_for(self, name) -> str:
|
|
source_count = len(self.sources.get(name, ()))
|
|
name_clean = re.sub(r"\W", "_", name)
|
|
return f"audit_event_{name_clean}_{source_count}"
|
|
|
|
def rows(self) -> Iterator[tuple[str, list[str], list[tuple[str, str]]]]:
|
|
for name in sorted(self.events.keys()):
|
|
yield name, self.events[name], self.sources[name]
|
|
|
|
|
|
def initialise_audit_events(app: Sphinx) -> None:
|
|
"""Initialise the audit_events attribute on the environment."""
|
|
if not hasattr(app.env, "audit_events"):
|
|
app.env.audit_events = AuditEvents()
|
|
|
|
|
|
def audit_events_purge(
|
|
app: Sphinx, env: BuildEnvironment, docname: str
|
|
) -> None:
|
|
"""This is to remove traces of removed documents from env.audit_events."""
|
|
fresh_audit_events = AuditEvents()
|
|
for name, args, (doc, target) in env.audit_events:
|
|
if doc != docname:
|
|
fresh_audit_events.add_event(name, args, (doc, target))
|
|
|
|
|
|
def audit_events_merge(
|
|
app: Sphinx,
|
|
env: BuildEnvironment,
|
|
docnames: list[str],
|
|
other: BuildEnvironment,
|
|
) -> None:
|
|
"""In Sphinx parallel builds, this merges audit_events from subprocesses."""
|
|
for name, args, source in other.audit_events:
|
|
env.audit_events.add_event(name, args, source)
|
|
|
|
|
|
class AuditEvent(SphinxDirective):
|
|
has_content = True
|
|
required_arguments = 1
|
|
optional_arguments = 2
|
|
final_argument_whitespace = True
|
|
|
|
_label = [
|
|
sphinx_gettext(
|
|
"Raises an :ref:`auditing event <auditing>` "
|
|
"{name} with no arguments."
|
|
),
|
|
sphinx_gettext(
|
|
"Raises an :ref:`auditing event <auditing>` "
|
|
"{name} with argument {args}."
|
|
),
|
|
sphinx_gettext(
|
|
"Raises an :ref:`auditing event <auditing>` "
|
|
"{name} with arguments {args}."
|
|
),
|
|
]
|
|
|
|
def run(self) -> list[nodes.paragraph]:
|
|
name = self.arguments[0]
|
|
if len(self.arguments) >= 2 and self.arguments[1]:
|
|
args = [
|
|
arg
|
|
for argument in self.arguments[1].strip("'\"").split(",")
|
|
if (arg := argument.strip())
|
|
]
|
|
else:
|
|
args = []
|
|
ids = []
|
|
try:
|
|
target = self.arguments[2].strip("\"'")
|
|
except (IndexError, TypeError):
|
|
target = None
|
|
if not target:
|
|
target = self.env.audit_events.id_for(name)
|
|
ids.append(target)
|
|
self.env.audit_events.add_event(name, args, (self.env.docname, target))
|
|
|
|
node = nodes.paragraph("", classes=["audit-hook"], ids=ids)
|
|
self.set_source_info(node)
|
|
if self.content:
|
|
node.rawsource = '\n'.join(self.content) # for gettext
|
|
self.state.nested_parse(self.content, self.content_offset, node)
|
|
else:
|
|
num_args = min(2, len(args))
|
|
text = self._label[num_args].format(
|
|
name=f"``{name}``",
|
|
args=", ".join(f"``{a}``" for a in args),
|
|
)
|
|
node.rawsource = text # for gettext
|
|
parsed, messages = self.state.inline_text(text, self.lineno)
|
|
node += parsed
|
|
node += messages
|
|
return [node]
|
|
|
|
|
|
class audit_event_list(nodes.General, nodes.Element): # noqa: N801
|
|
pass
|
|
|
|
|
|
class AuditEventListDirective(SphinxDirective):
|
|
def run(self) -> list[audit_event_list]:
|
|
return [audit_event_list()]
|
|
|
|
|
|
class AuditEventListTransform(SphinxPostTransform):
|
|
default_priority = 500
|
|
|
|
def run(self) -> None:
|
|
if self.document.next_node(audit_event_list) is None:
|
|
return
|
|
|
|
table = self._make_table(self.app.builder, self.env.docname)
|
|
for node in self.document.findall(audit_event_list):
|
|
node.replace_self(table)
|
|
|
|
def _make_table(self, builder: Builder, docname: str) -> nodes.table:
|
|
table = nodes.table(cols=3)
|
|
group = nodes.tgroup(
|
|
"",
|
|
nodes.colspec(colwidth=30),
|
|
nodes.colspec(colwidth=55),
|
|
nodes.colspec(colwidth=15),
|
|
cols=3,
|
|
)
|
|
head = nodes.thead()
|
|
body = nodes.tbody()
|
|
|
|
table += group
|
|
group += head
|
|
group += body
|
|
|
|
head += nodes.row(
|
|
"",
|
|
nodes.entry("", nodes.paragraph("", "Audit event")),
|
|
nodes.entry("", nodes.paragraph("", "Arguments")),
|
|
nodes.entry("", nodes.paragraph("", "References")),
|
|
)
|
|
|
|
for name, args, sources in builder.env.audit_events.rows():
|
|
body += self._make_row(builder, docname, name, args, sources)
|
|
|
|
return table
|
|
|
|
@staticmethod
|
|
def _make_row(
|
|
builder: Builder,
|
|
docname: str,
|
|
name: str,
|
|
args: list[str],
|
|
sources: list[tuple[str, str]],
|
|
) -> nodes.row:
|
|
row = nodes.row()
|
|
name_node = nodes.paragraph("", nodes.Text(name))
|
|
row += nodes.entry("", name_node)
|
|
|
|
args_node = nodes.paragraph()
|
|
for arg in args:
|
|
args_node += nodes.literal(arg, arg)
|
|
args_node += nodes.Text(", ")
|
|
if len(args_node.children) > 0:
|
|
args_node.children.pop() # remove trailing comma
|
|
row += nodes.entry("", args_node)
|
|
|
|
backlinks_node = nodes.paragraph()
|
|
backlinks = enumerate(sorted(set(sources)), start=1)
|
|
for i, (doc, label) in backlinks:
|
|
if isinstance(label, str):
|
|
ref = nodes.reference("", f"[{i}]", internal=True)
|
|
try:
|
|
target = (
|
|
f"{builder.get_relative_uri(docname, doc)}#{label}"
|
|
)
|
|
except NoUri:
|
|
continue
|
|
else:
|
|
ref["refuri"] = target
|
|
backlinks_node += ref
|
|
row += nodes.entry("", backlinks_node)
|
|
return row
|
|
|
|
|
|
def setup(app: Sphinx):
|
|
app.add_directive("audit-event", AuditEvent)
|
|
app.add_directive("audit-event-table", AuditEventListDirective)
|
|
app.add_post_transform(AuditEventListTransform)
|
|
app.connect("builder-inited", initialise_audit_events)
|
|
app.connect("env-purge-doc", audit_events_purge)
|
|
app.connect("env-merge-info", audit_events_merge)
|
|
return {
|
|
"version": "1.0",
|
|
"parallel_read_safe": True,
|
|
"parallel_write_safe": True,
|
|
}
|