diff --git a/contracts/scripted-gameplay-audit/v1.json b/contracts/scripted-gameplay-audit/v1.json new file mode 100644 index 000000000..74adc508c --- /dev/null +++ b/contracts/scripted-gameplay-audit/v1.json @@ -0,0 +1,624 @@ +{ + "schema_version": 1, + "source_contexts": [ + { + "path": "maps/python/Auction.py", + "scope": "item_buy", + "context_sha256": "4d1c248a03f7d321a01264312f654825945c78b05873e68772bb789755b50064" + }, + { + "path": "maps/python/Guild.py", + "scope": "Guild.log_add", + "context_sha256": "15b40c1c01de41badb530eaa91907895b2d00f37a8f1e4d617a6c734d8eea9f4" + }, + { + "path": "maps/python/Interface.py", + "scope": "Interface.dbg", + "context_sha256": "21417ab27059ae9210c757b38290553719e348d5d10822ed37ecd8a56679f261" + }, + { + "path": "maps/python/Jail.py", + "scope": "Jail.jail", + "context_sha256": "e3922f75bc55c572f22ac62a4d058116945d6b15e18452ae8a1e52cb776868f3" + }, + { + "path": "maps/python/Merchant.py", + "scope": "Seller.dialog", + "context_sha256": "9770db477756a28445f26026696932b7a42ce0425e59e86f5b6889a633d6418f" + }, + { + "path": "maps/python/Merchant.py", + "scope": "SpellSeller.dialog", + "context_sha256": "ade70420a60bfd2e697cec35ce4fc7c9ffca95841f357412296dd3b0a2367ec9" + }, + { + "path": "maps/python/PostOffice.py", + "scope": "PostOffice._withdraw_one", + "context_sha256": "8d3cdbef62c27714268c2af54a71eac8e8c79bc02937d3325178894a2eed9bbe" + }, + { + "path": "maps/python/QuestManager.py", + "scope": "QuestManager.complete", + "context_sha256": "fe39cddae92a37e0e3a488585cd93ae952a1a1e26b2f89309d799489bbe21ae6" + }, + { + "path": "maps/python/Thieves.py", + "scope": "clear_bounty", + "context_sha256": "a9b32636d482c49a13beacc45eb5c168ecbe0fb52d0b0988650c5445c2fb1cab" + }, + { + "path": "maps/python/auctions/clerk.py", + "scope": "main", + "context_sha256": "be7b9f2cfa308032e2b8a58a0271ccc0b86e98dced524726d9fa16d655cfb3af" + }, + { + "path": "maps/python/commands/console.py", + "scope": "AutoComplete._assignment", + "context_sha256": "0e603d575923c09f32f88f38968b1d34dd4760cfb6d250f42bc00c06b25c42ef" + }, + { + "path": "maps/python/commands/console.py", + "scope": "PyConsole.push", + "context_sha256": "05655ddeaac50117dbb0f026a6d4b8f098deddc66cda4073c1840b0f7d150721" + }, + { + "path": "maps/python/commands/console.py", + "scope": "main", + "context_sha256": "dfb065fe83c3c5830324949de4d1b0cd008b0587e1244a0dbfb0551cbe420ab9" + }, + { + "path": "maps/python/commands/guild.py", + "scope": "main", + "context_sha256": "81b7581b8f472245ea4c0a8a365ce8ded96f77aa674e7eb55537ef0374e3c539" + }, + { + "path": "maps/python/events/init/documentation.py", + "scope": "dump_docstring", + "context_sha256": "71b799f5d3a46c7bd83415c588d8185b086b602831b33ebe8f3016096022fc9c" + }, + { + "path": "maps/python/events/init/documentation.py", + "scope": "dump_obj", + "context_sha256": "c3f38e08b4b388f924a6ae25e2fa6ed100a42d0b0bce57c307b168fbeb6e9b9d" + }, + { + "path": "maps/python/events/init/documentation.py", + "scope": "getargspec", + "context_sha256": "92f6cad6280fa2d3a4e955633f3858ce670d5ed56ebc8c3b71c1f40e388fd25e" + }, + { + "path": "maps/python/events/python_exception.py", + "scope": "", + "context_sha256": "b0208223d2acee5b5ddea7fb793a077de563fe87b4390bba715833707af7d2db" + }, + { + "path": "maps/python/events/python_init.py", + "scope": "", + "context_sha256": "dc2d716792181eb32d36c84e83b281423bd72f824e76bf6a6179b05268eea4e2" + }, + { + "path": "maps/python/events/python_print.py", + "scope": "", + "context_sha256": "31c036d294a46d9801e6f48935230a66655442fc052c1f28dea34e6f0efe6061" + }, + { + "path": "maps/python/generic/apartment_seller.py", + "scope": "main", + "context_sha256": "007f4e0263e7d42e362fb222d2401849516598f20ed37df4faa2f0c760d010b8" + }, + { + "path": "maps/python/generic/postoffice_clerk.py", + "scope": "main", + "context_sha256": "9a7bee3014e61991225e521b86d1e3ba463439a5568ecf8b1f4d6ba0d9e0edcb" + }, + { + "path": "maps/python/guilds/guild.py", + "scope": "main", + "context_sha256": "bb968b0f55372f2e91e4cbbaca7b631b74350b291781371f7088469c90dd9a66" + }, + { + "path": "maps/python/guilds/storage.py", + "scope": "main", + "context_sha256": "07d4b6f8a12104f9cba172e7e94126d8b1a2fc0ce3c3d2ee276234724de13750" + }, + { + "path": "maps/python/houses/fees_checker.py", + "scope": "main", + "context_sha256": "24fe8a6198208f94f3816abb81216551a33cd4a7356b3ef54a797f445690d088" + }, + { + "path": "maps/python/houses/manager.py", + "scope": "main", + "context_sha256": "d76efa6fb916a6a1c392afd6f0f00826895cbbd509b8d1b52081e4d0fa6e5a85" + }, + { + "path": "maps/shattered_islands/strakewood_island/underground_city/scripts/torches.py", + "scope": "main", + "context_sha256": "7b56a0e80b8121fb3590108c49db695080e02601565120d8a324e48db8609efb" + }, + { + "path": "maps/shattered_islands/strakewood_island/underground_city/scripts/torches.py", + "scope": "timer", + "context_sha256": "c7631e25ca659448db9592d625a84faa4f6e5a2ab991e990911634794b8cc0ee" + } + ], + "metric_sites": [ + { + "path": "maps/python/Auction.py", + "ast_path": "body[11].body[15].value", + "scope": "item_buy", + "method": "MetricAdd", + "metric": "economy.auction_purchases", + "classification": "gameplay-journal", + "proposed_journal_reason": "auction.purchase", + "event_rate": "One per successful player auction purchase.", + "rationale": "The debit, delivery, and payout need one recovery-grade transaction; this current post-payment metric site must move to that transaction's terminal success boundary." + }, + { + "path": "maps/python/Auction.py", + "ast_path": "body[11].body[16].value", + "scope": "item_buy", + "method": "MetricAdd", + "metric": "economy.auction_currency_spent", + "classification": "gameplay-journal", + "proposed_journal_reason": "auction.purchase", + "event_rate": "One per successful player auction purchase.", + "rationale": "This aggregate must advance exactly once only after the correlated debit, delivery, and payout commit." + }, + { + "path": "maps/python/Jail.py", + "ast_path": "body[6].body[2].body[11].value", + "scope": "Jail.jail", + "method": "MetricAdd", + "metric": "social.jail_sentences", + "classification": "gameplay-journal", + "proposed_journal_reason": "social.jail-placement", + "event_rate": "One per successful jail placement.", + "rationale": "Location, savebed, and sentence state form a rare support-relevant transition; the current site precedes all terminal player feedback." + }, + { + "path": "maps/python/Jail.py", + "ast_path": "body[6].body[2].body[12].body[0].value", + "scope": "Jail.jail", + "method": "MetricAdd", + "metric": "social.jail_time_sentenced", + "classification": "gameplay-journal", + "proposed_journal_reason": "social.jail-placement", + "event_rate": "One bounded duration value per successful non-lifetime sentence.", + "rationale": "The duration is typed context for the same placement, not an independent event." + }, + { + "path": "maps/python/Merchant.py", + "ast_path": "body[10].body[13].body[10].value", + "scope": "SpellSeller.dialog", + "method": "MetricAdd", + "metric": "economy.shop_purchases", + "classification": "gameplay-journal", + "proposed_journal_reason": "merchant.purchase", + "event_rate": "One per successful scripted spell purchase.", + "rationale": "This current pre-delivery metric can survive failed spell creation and must move behind an idempotent payment-and-learning commit." + }, + { + "path": "maps/python/Merchant.py", + "ast_path": "body[10].body[13].body[11].value", + "scope": "SpellSeller.dialog", + "method": "MetricAdd", + "metric": "economy.shop_currency_spent", + "classification": "gameplay-journal", + "proposed_journal_reason": "merchant.purchase", + "event_rate": "One per successful scripted spell purchase.", + "rationale": "The aggregate must advance once only after payment and spell creation commit together." + }, + { + "path": "maps/python/Merchant.py", + "ast_path": "body[8].body[17].body[9].body[3].orelse[0].orelse[0].value", + "scope": "Seller.dialog", + "method": "MetricAdd", + "metric": "economy.shop_purchases", + "classification": "gameplay-journal", + "proposed_journal_reason": "merchant.purchase", + "event_rate": "One per successful scripted item purchase.", + "rationale": "This current pre-clone/pre-delivery metric can record an incomplete purchase and must move behind the correlated commit." + }, + { + "path": "maps/python/Merchant.py", + "ast_path": "body[8].body[17].body[9].body[3].orelse[0].orelse[1].value", + "scope": "Seller.dialog", + "method": "MetricAdd", + "metric": "economy.shop_currency_spent", + "classification": "gameplay-journal", + "proposed_journal_reason": "merchant.purchase", + "event_rate": "One per successful scripted item purchase.", + "rationale": "The aggregate must advance once only after payment, delivery, and stock adjustment commit together." + }, + { + "path": "maps/python/PostOffice.py", + "ast_path": "body[2].body[8].body[3].value", + "scope": "PostOffice._withdraw_one", + "method": "MetricAdd", + "metric": "social.post_items_received", + "classification": "gameplay-journal", + "proposed_journal_reason": "post.receive", + "event_rate": "One per successfully delivered queued post item.", + "rationale": "The current site precedes durable queue removal; delivery, removal, and the aggregate need exact-once reconciliation." + }, + { + "path": "maps/python/QuestManager.py", + "ast_path": "body[4].body[26].body[12].value", + "scope": "QuestManager.complete", + "method": "MetricMarkUnique", + "metric": "quests.completed_parts", + "classification": "gameplay-journal", + "proposed_journal_reason": "quest.part-completed", + "event_rate": "At most one unique mark per qualified quest part and character.", + "rationale": "This aggregate projects the same stable quest-part lifecycle intent after authoritative state and objective updates; terminal journal uncertainty remains pending for reconciliation." + }, + { + "path": "maps/python/QuestManager.py", + "ast_path": "body[4].body[26].body[13].value", + "scope": "QuestManager.complete", + "method": "MetricKeyedAdd", + "metric": "quests.part_completions_by_id", + "classification": "gameplay-journal", + "proposed_journal_reason": "quest.part-completed", + "event_rate": "One per successful top-level or nested quest-part completion.", + "rationale": "Stable qualified part identity is shared with the durable quest.part-completed transition, never each intermediate state write." + }, + { + "path": "maps/python/QuestManager.py", + "ast_path": "body[4].body[26].body[20].body[0].value", + "scope": "QuestManager.complete", + "method": "MetricAdd", + "metric": "quests.repeatable_completed", + "classification": "gameplay-journal", + "proposed_journal_reason": "quest.completed", + "event_rate": "One per completed repeatable quest cycle.", + "rationale": "The durable quest.completed transition identifies the terminal repeat cycle; intermediate state writes remain unrecorded." + }, + { + "path": "maps/python/Thieves.py", + "ast_path": "body[2].body[3].value", + "scope": "clear_bounty", + "method": "MetricAdd", + "metric": "social.bounties_cleared", + "classification": "gameplay-journal", + "proposed_journal_reason": "social.bounty-clear", + "event_rate": "One per successfully paid bounty clearance.", + "rationale": "The rare persistent faction transition has support value but must commit with its payment." + }, + { + "path": "maps/python/Thieves.py", + "ast_path": "body[2].body[4].value", + "scope": "clear_bounty", + "method": "MetricAdd", + "metric": "economy.bounty_currency_spent", + "classification": "gameplay-journal", + "proposed_journal_reason": "social.bounty-clear", + "event_rate": "One per successfully paid bounty clearance.", + "rationale": "Payment and faction-state clearance must be one terminal business outcome." + }, + { + "path": "maps/python/Thieves.py", + "ast_path": "body[2].body[5].value", + "scope": "clear_bounty", + "method": "MetricKeyedAdd", + "metric": "social.bounties_cleared_by_faction", + "classification": "gameplay-journal", + "proposed_journal_reason": "social.bounty-clear", + "event_rate": "One stable faction key per successfully cleared bounty.", + "rationale": "Faction identity is typed context for the same transaction, not an independent event." + }, + { + "path": "maps/python/auctions/clerk.py", + "ast_path": "body[16].body[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].body[7].body[15].value", + "scope": "main", + "method": "MetricAdd", + "metric": "economy.auction_listings", + "classification": "gameplay-journal", + "proposed_journal_reason": "auction.list", + "event_rate": "One per successfully persisted auction listing.", + "rationale": "Player custody, listing identity, UID advancement, and map placement must reconcile before this metric commits." + }, + { + "path": "maps/python/generic/apartment_seller.py", + "ast_path": "body[4].body[2].orelse[0].orelse[0].body[4].body[1].value", + "scope": "main", + "method": "MetricAdd", + "metric": "economy.housing_purchases", + "classification": "gameplay-journal", + "proposed_journal_reason": "housing.purchase", + "event_rate": "One per successful apartment purchase or upgrade.", + "rationale": "This current post-debit/pre-ownership metric must move behind payment, ownership, and apartment-swap commit." + }, + { + "path": "maps/python/generic/apartment_seller.py", + "ast_path": "body[4].body[2].orelse[0].orelse[0].body[4].body[2].value", + "scope": "main", + "method": "MetricAdd", + "metric": "economy.housing_currency_spent", + "classification": "gameplay-journal", + "proposed_journal_reason": "housing.purchase", + "event_rate": "One per successful apartment purchase or upgrade.", + "rationale": "The aggregate must advance once only after every persistent apartment effect commits." + }, + { + "path": "maps/python/generic/postoffice_clerk.py", + "ast_path": "body[5].body[0].orelse[0].orelse[0].body[5].body[2].value", + "scope": "main", + "method": "MetricAdd", + "metric": "social.post_items_sent", + "classification": "gameplay-journal", + "proposed_journal_reason": "post.send", + "event_rate": "One per successfully paid and queued post item.", + "rationale": "Payment, serialized delivery, source removal, and this aggregate need one exact-once transaction." + }, + { + "path": "maps/python/generic/postoffice_clerk.py", + "ast_path": "body[5].body[0].orelse[0].orelse[0].body[5].body[3].value", + "scope": "main", + "method": "MetricAdd", + "metric": "economy.postage_currency_spent", + "classification": "gameplay-journal", + "proposed_journal_reason": "post.send", + "event_rate": "One per successfully paid and queued post item.", + "rationale": "The aggregate must be terminal evidence for the same correlated send, not merely proof of debit." + }, + { + "path": "maps/python/guilds/guild.py", + "ast_path": "body[5].body[0].orelse[0].orelse[0].body[3].orelse[0].value", + "scope": "main", + "method": "MetricAdd", + "metric": "social.guild_departures", + "classification": "gameplay-journal", + "proposed_journal_reason": "social.guild-departure", + "event_rate": "One per successful approved-member departure.", + "rationale": "A rare persistent membership-state transition is useful for support and recovery." + }, + { + "path": "maps/python/guilds/guild.py", + "ast_path": "body[5].body[0].orelse[0].orelse[0].orelse[0].body[1].orelse[1].value", + "scope": "main", + "method": "MetricAdd", + "metric": "social.guild_applications", + "classification": "gameplay-journal", + "proposed_journal_reason": "social.guild-application", + "event_rate": "One per successful guild application creation.", + "rationale": "A rare persistent membership-state transition is useful for support and recovery." + }, + { + "path": "maps/python/houses/fees_checker.py", + "ast_path": "body[2].body[2].body[1].body[0].value", + "scope": "main", + "method": "MetricAdd", + "metric": "economy.housing_fees_paid", + "classification": "gameplay-journal", + "proposed_journal_reason": "housing.fee-payment", + "event_rate": "At most one per house-entry fee renewal.", + "rationale": "This current post-debit/pre-expiration metric must move behind the correlated payment and access-state commit." + }, + { + "path": "maps/python/houses/fees_checker.py", + "ast_path": "body[2].body[2].body[1].body[1].value", + "scope": "main", + "method": "MetricAdd", + "metric": "economy.housing_currency_spent", + "classification": "gameplay-journal", + "proposed_journal_reason": "housing.fee-payment", + "event_rate": "At most one per house-entry fee renewal.", + "rationale": "The aggregate must advance once only after fee state is durable." + }, + { + "path": "maps/python/houses/manager.py", + "ast_path": "body[5].body[0].orelse[0].orelse[0].orelse[0].orelse[0].body[0].orelse[0].body[0].body[0].value", + "scope": "main", + "method": "MetricAdd", + "metric": "economy.housing_purchases", + "classification": "gameplay-journal", + "proposed_journal_reason": "housing.purchase", + "event_rate": "One per successful house purchase.", + "rationale": "This current post-debit/pre-ownership metric must move behind the correlated ownership commit." + }, + { + "path": "maps/python/houses/manager.py", + "ast_path": "body[5].body[0].orelse[0].orelse[0].orelse[0].orelse[0].body[0].orelse[0].body[0].body[1].value", + "scope": "main", + "method": "MetricAdd", + "metric": "economy.housing_currency_spent", + "classification": "gameplay-journal", + "proposed_journal_reason": "housing.purchase", + "event_rate": "One per successful house purchase.", + "rationale": "The aggregate must advance once only after payment and ownership commit together." + } + ], + "audit_like_sites": [ + { + "path": "maps/python/Guild.py", + "ast_path": "body[5].body[49]", + "scope": "Guild.log_add", + "facility": "Guild.log_add sink", + "classification": "operational/security-log", + "event_rate": "One protected file append per accepted guild audit message.", + "rationale": "This is the human-text persistence sink; path construction, append behavior, and arbitrary text must remain outside structured gameplay records." + }, + { + "path": "maps/python/Interface.py", + "ast_path": "body[5].body[2].body[1].value", + "scope": "Interface.dbg", + "facility": "Atrinik.Logger", + "classification": "operational/security-log", + "event_rate": "Only when interface debug mode is explicitly enabled.", + "rationale": "Formatted developer diagnostics are human operational text and never structured gameplay evidence." + }, + { + "path": "maps/python/auctions/clerk.py", + "ast_path": "body[16].body[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].body[4].body[0].value", + "scope": "main", + "facility": "Python.print", + "classification": "operational/security-log", + "event_rate": "At most one diagnostic when an auction-house sign is missing during clerk initialization.", + "rationale": "Map-configuration diagnostics are routed to the protected Python log and are not gameplay evidence." + }, + { + "path": "maps/python/commands/console.py", + "ast_path": "body[11].body[4].body[2].body[0].value", + "scope": "AutoComplete._assignment", + "facility": "Python.eval", + "classification": "operational/security-log", + "event_rate": "One expression evaluation per privileged console autocomplete assignment probe.", + "rationale": "Dynamic operator-console evaluation is an explicit protected execution boundary and must never become structured gameplay evidence." + }, + { + "path": "maps/python/commands/console.py", + "ast_path": "body[13].body[3].body[6].body[0].value", + "scope": "PyConsole.push", + "facility": "Python.InteractiveConsole.push", + "classification": "operational/security-log", + "event_rate": "One execution attempt per privileged console submission after multiline buffering.", + "rationale": "The interactive interpreter executes arbitrary operator-supplied Python and is a protected execution boundary, never structured gameplay evidence." + }, + { + "path": "maps/python/commands/console.py", + "ast_path": "body[18].body[2].value", + "scope": "main", + "facility": "Logger", + "classification": "operational/security-log", + "event_rate": "One protected log entry per privileged console submission.", + "rationale": "The entry deliberately contains arbitrary operator-supplied Python and must never enter structured gameplay records." + }, + { + "path": "maps/python/commands/guild.py", + "ast_path": "body[4].body[6].value", + "scope": "main", + "facility": "Logger", + "classification": "operational/security-log", + "event_rate": "One human log entry per accepted guild-chat message.", + "rationale": "Player-authored chat is intentionally human operational text and prohibited from structured gameplay records." + }, + { + "path": "maps/python/events/init/documentation.py", + "ast_path": "body[13].body[2].body[1].value", + "scope": "getargspec", + "facility": "Python.print", + "classification": "operational/security-log", + "event_rate": "One diagnostic per undocumented callable signature encountered during documentation generation.", + "rationale": "Developer-facing introspection text may contain object representations and remains operational only." + }, + { + "path": "maps/python/events/init/documentation.py", + "ast_path": "body[14].body[3].body[2].body[0].value", + "scope": "dump_docstring", + "facility": "Python.print", + "classification": "operational/security-log", + "event_rate": "One diagnostic per malformed generated property documentation entry.", + "rationale": "Documentation diagnostics are implementation text, not structured gameplay evidence." + }, + { + "path": "maps/python/events/init/documentation.py", + "ast_path": "body[14].body[3].body[4].body[0].value", + "scope": "dump_docstring", + "facility": "Python.print", + "classification": "operational/security-log", + "event_rate": "One diagnostic per generated property whose documented type cannot be parsed.", + "rationale": "Documentation diagnostics are implementation text, not structured gameplay evidence." + }, + { + "path": "maps/python/events/init/documentation.py", + "ast_path": "body[16].body[3].body[3].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[8].body[0].value", + "scope": "dump_obj", + "facility": "Python.print", + "classification": "operational/security-log", + "event_rate": "One diagnostic per undocumented constant encountered during documentation generation.", + "rationale": "Generated-documentation warnings remain protected operational text." + }, + { + "path": "maps/python/events/python_exception.py", + "ast_path": "body[4].body[0].body[0].value", + "scope": "", + "facility": "Atrinik.Logger", + "classification": "operational/security-log", + "event_rate": "One line per uncaught Python exception traceback line.", + "rationale": "Runtime diagnostics may contain implementation details and belong only in the protected operational log." + }, + { + "path": "maps/python/events/python_init.py", + "ast_path": "body[7].body[1].body[0].body[0].body[0].value", + "scope": "", + "facility": "Python.exec", + "classification": "operational/security-log", + "event_rate": "One execution per configured authored Python initialization file at server startup.", + "rationale": "Authored bootstrap execution is a privileged operational boundary; its source is not a script-formatted gameplay record." + }, + { + "path": "maps/python/events/python_print.py", + "ast_path": "body[2].body[0].body[0].value", + "scope": "", + "facility": "Atrinik.Logger", + "classification": "operational/security-log", + "event_rate": "One line per explicit Python print event.", + "rationale": "Free-form script diagnostics are operational text and never authoritative structured gameplay evidence." + }, + { + "path": "maps/python/guilds/storage.py", + "ast_path": "body[7].body[0].orelse[0].body[2].value", + "scope": "main", + "facility": "Guild.log_add", + "classification": "operational/security-log", + "event_rate": "One entry per permitted guild-storage floor drop.", + "rationale": "Display names remain human access evidence; typed item custody must come from the server move boundary." + }, + { + "path": "maps/python/guilds/storage.py", + "ast_path": "body[7].body[0].orelse[0].orelse[0].body[2].value", + "scope": "main", + "facility": "Guild.log_add", + "classification": "operational/security-log", + "event_rate": "One entry per permitted guild-storage container insertion.", + "rationale": "Display names remain human access evidence; typed item custody must come from the server move boundary." + }, + { + "path": "maps/python/guilds/storage.py", + "ast_path": "body[7].body[0].orelse[0].orelse[0].orelse[0].body[2].body[0].value", + "scope": "main", + "facility": "Guild.log_add", + "classification": "operational/security-log", + "event_rate": "One entry per permitted pickup from a guild-storage container.", + "rationale": "Container and item display names must not be copied into structured records." + }, + { + "path": "maps/python/guilds/storage.py", + "ast_path": "body[7].body[0].orelse[0].orelse[0].orelse[0].body[2].orelse[0].value", + "scope": "main", + "facility": "Guild.log_add", + "classification": "operational/security-log", + "event_rate": "One entry per permitted pickup from the guild-storage floor.", + "rationale": "Item display names remain human access evidence, while server custody events use stable typed identity." + }, + { + "path": "maps/python/guilds/storage.py", + "ast_path": "body[7].body[0].orelse[0].orelse[0].orelse[0].orelse[0].orelse[0].body[1].value", + "scope": "main", + "facility": "Guild.log_add", + "classification": "operational/security-log", + "event_rate": "One entry per permitted privileged bulk drop/take command.", + "rationale": "The message includes arbitrary command text and is categorically prohibited from structured gameplay records." + }, + { + "path": "maps/shattered_islands/strakewood_island/underground_city/scripts/torches.py", + "ast_path": "body[1].body[3].value", + "scope": "timer", + "facility": "Atrinik.Eval", + "classification": "operational/security-log", + "event_rate": "One delayed callback evaluation per active underground-city torch sequence step.", + "rationale": "The fixed authored callback expression is a privileged engine execution boundary, not a structured gameplay record." + }, + { + "path": "maps/shattered_islands/strakewood_island/underground_city/scripts/torches.py", + "ast_path": "body[2].body[9].value", + "scope": "main", + "facility": "Atrinik.Eval", + "classification": "operational/security-log", + "event_rate": "One initial delayed callback evaluation when the torch sequence is activated.", + "rationale": "The fixed authored callback expression schedules operational script execution and carries no authoritative gameplay payload." + } + ] +} diff --git a/docs/SCRIPTED_GAMEPLAY_AUDIT.md b/docs/SCRIPTED_GAMEPLAY_AUDIT.md new file mode 100644 index 000000000..ea539b0e1 --- /dev/null +++ b/docs/SCRIPTED_GAMEPLAY_AUDIT.md @@ -0,0 +1,66 @@ +# Scripted gameplay audit boundaries + +`contracts/scripted-gameplay-audit/v1.json` is the reviewed occurrence inventory +of every authored Python gameplay metric and audit-like logging call. +The aggregate validator parses all `maps/**/*.py` source except test fixtures, +rejects dynamic or indirect reserved telemetry access, inventories privileged +dynamic execution boundaries, and binds every call to its source, +lexical scope, AST location, and a normalized hash of the surrounding function +or module. Moving, adding, removing, or semantically surrounding a site +therefore requires an explicit noise, privacy, and recovery decision. + +This is a governance check for trusted reviewed source, not a Python sandbox or +a proof over arbitrary metaprogramming. Authored telemetry must use the direct +forms represented by the contract; synthesizing it through imports, reflection, +namespaces, dynamic execution, or callable aliases is prohibited. The validator +rejects those reserved forms and fails closed when the direct-call inventory or +its surrounding context drifts. Runtime authorization and operator-console +security remain server responsibilities. +The telemetry spellings `Metric*`, `Logger`, `print`, and `log_add` are reserved +within authored maps; ambiguous shadowing, rebinding, or reflection is rejected. + +The four dispositions describe the intended telemetry boundary: + +- `gameplay-journal` is a bounded semantic transition useful for support or + recovery. Its proposed reason is ASCII, bounded to the Classic server's + 255-character gameplay-journal identifier limit, and contains no player + text. The Classic server contract is authoritative; quest lifecycle producers + now use it, while economy producers remain gated on the composition API. +- `aggregate-only` retains bounded statistics without ordered event evidence. +- `operational/security-log` remains protected human/operator diagnostics. +- `not-recorded` is neither useful nor appropriate to retain. + +All 26 current metric calls are low-volume quest, post, auction, merchant, +housing, bounty, guild, or jail outcomes classified as gameplay-journal +projections. Quest lifecycle producers now use the stable Classic contract. +This classification does not claim that the remaining legacy economy placement +is transactionally safe. Merchant purchase metrics currently +precede item or spell delivery, post collection precedes queue removal, and +housing metrics can follow debit while preceding the ownership or fee update. +They must move behind the durable idempotent commit/reconciliation result when +the typed APIs become available; generic payment or custody hooks must not add +a second copy of a business-specific aggregate. + +The 21 current audit-like sites include generic Python diagnostics and prints, +the privileged `eval`, `exec`, engine callback, and interactive-console execution boundaries, +guild chat and console commands, guild-storage `Guild.log_add` calls, and their +human-text file sink. Some carry +display names or arbitrary operator/player text. That text must not enter +structured gameplay records. Successful storage custody instead belongs to the +server's typed item transaction at the authoritative post-veto move boundary. + +High-volume movement, traversal, attacks, ordinary kills, damage, healing, +regeneration, routine spell/skill/consumable use, emotes, and every intermediate +quest-state write have no scripted metric sites and remain outside the +structured gameplay journal. Classified guild chat remains only in its +protected operational log. Adding a metric site requires a reviewed contract +row; it never becomes a journal producer merely because a metric is useful. + +The stable quest contract from https://github.com/atrinik/classic/issues/161 is +integrated by the shared `QuestManager`. Crash-safe quest replay still requires +the server-owned idempotency and disposition contract tracked by +https://github.com/atrinik/classic/issues/321. Remaining executable economy +integration depends on the scripted multi-step composition contract in +https://github.com/atrinik/classic/issues/313. Content must use those typed APIs +rather than append raw logs, invent a second audit store, or misuse one journal +kind for unrelated item and currency flows. diff --git a/maps/python/tests/QuestManager.py b/maps/python/tests/QuestManager.py index 8d77ef299..0ed5e5162 100644 --- a/maps/python/tests/QuestManager.py +++ b/maps/python/tests/QuestManager.py @@ -509,6 +509,7 @@ def journal_begin(*args): mock.patch.object( qm, "journal_commit", side_effect=commits.append): qm.reset_quest() + qm.reset_quest() self.assertFalse(qm.started()) self.assertEqual([ ("quest.repeat-reset", "quest:repeat_failure_test_quest", @@ -678,6 +679,83 @@ def test_14_kept_objective_item_is_retained_without_quest_flags(self): self.assertFalse(sword.f_startequip) sword.Destroy("test.quest-objective-cleanup") + def test_15_failed_journal_hooks_are_exactly_once(self): + quest = { + "parts": OrderedDict((("attempt", { + "info": "", + "uid": "attempt", + "name": "Attempt", + }),)), + "name": "Failed Journal Hook Quest", + "uid": "failed_journal_hook_quest", + } + qm = QuestManager(activator, quest) + qm.start("attempt") + intents = [] + commits = [] + + def journal_begin(reason, subject, before, after, lineage=""): + transaction = "failure-{}".format(len(intents)) + intents.append((reason, subject, before, after, lineage)) + return transaction + + with mock.patch.object(qm, "journal_begin", side_effect=journal_begin), \ + mock.patch.object( + qm, "journal_commit", side_effect=commits.append): + self.assertTrue(qm.fail("attempt")) + self.assertFalse(qm.fail("attempt")) + + self.assertEqual([ + ("quest.part-failed", + "quest-part:failed_journal_hook_quest::attempt", + Atrinik.QUEST_STATUS_STARTED, Atrinik.QUEST_STATUS_FAILED, ""), + ("quest.failed", "quest:failed_journal_hook_quest", + Atrinik.QUEST_STATUS_STARTED, Atrinik.QUEST_STATUS_FAILED, ""), + ], intents) + self.assertEqual(["failure-0", "failure-1"], commits) + + def test_16_objective_removal_uses_one_typed_decrease_per_stack(self): + qm = object.__new__(QuestManager) + qm.activator = mock.Mock() + first = mock.Mock(nrof=2) + second = mock.Mock(nrof=5) + qm.activator.FindObjects.return_value = [first, second] + objective = mock.Mock(sub_type=Atrinik.QUEST_TYPE_ITEM) + + qm.remove_quest_items({ + "item": { + "arch": "sword", + "name": "objective sword", + "nrof": 4, + }, + }, objective) + + first.Decrease.assert_called_once_with( + 2, reason="quest.objective-remove" + ) + second.Decrease.assert_called_once_with( + 2, reason="quest.objective-remove" + ) + + def test_17_rejected_intent_precedes_start_mutation(self): + quest = { + "parts": OrderedDict((("attempt", { + "info": "", + "uid": "attempt", + "name": "Attempt", + }),)), + "name": "Rejected Intent Quest", + "uid": "rejected_intent_quest", + } + qm = QuestManager(activator, quest) + + with mock.patch.object( + qm, "journal_begin", + side_effect=RuntimeError("forced intent rejection")): + with self.assertRaisesRegex(RuntimeError, "forced intent"): + qm.start("attempt") + + self.assertFalse(qm.started()) activator = Atrinik.WhoIsActivator() me = Atrinik.WhoAmI() diff --git a/tools/scripted_gameplay_audit.py b/tools/scripted_gameplay_audit.py new file mode 100644 index 000000000..9787da71c --- /dev/null +++ b/tools/scripted_gameplay_audit.py @@ -0,0 +1,1111 @@ +#!/usr/bin/env python3 +"""Validate the reviewed disposition of authored gameplay telemetry sites.""" + +from __future__ import annotations + +import ast +import hashlib +import json +from pathlib import Path, PurePosixPath +import re +from typing import Any, Iterator + + +MANIFEST_PATH = Path("contracts/scripted-gameplay-audit/v1.json") +METRIC_METHODS = frozenset({"MetricAdd", "MetricKeyedAdd", "MetricMarkUnique"}) +CLASSIFICATIONS = frozenset( + {"gameplay-journal", "aggregate-only", "operational/security-log", "not-recorded"} +) +METRIC_ID = re.compile(r"^[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)+$") +JOURNAL_REASON = re.compile(r"^[a-z][a-z0-9]*(?:[.-][a-z0-9]+)*$") +IDENTITY_MAX = 255 +SENSITIVE_RECEIVER_NAMES = frozenset( + {"Atrinik", "activator", "controller", "guild", "pl", "player"} +) +SENSITIVE_REFLECTIVE_NAMES = frozenset( + (*METRIC_METHODS, "Eval", "Logger", "log_add", "print", "__getattribute__") +) +DYNAMIC_EXECUTION_NAMES = frozenset({"compile", "eval", "exec", "__import__"}) +NAMESPACE_REFLECTION_NAMES = frozenset({"globals", "locals"}) + + +class ScriptedGameplayAuditError(ValueError): + """The authored scripted-gameplay audit is incomplete or malformed.""" + + +def _safe_relative_path(root: Path, value: object) -> str: + if not isinstance(value, str): + raise ScriptedGameplayAuditError("audit paths must be strings") + if "\\" in value or "\0" in value: + raise ScriptedGameplayAuditError("audit path is not canonical POSIX: {!r}".format(value)) + path = PurePosixPath(value) + if path.is_absolute() or not path.parts or "." in path.parts or ".." in path.parts: + raise ScriptedGameplayAuditError("unsafe audit path: {!r}".format(value)) + if not value.startswith("maps/") or not value.endswith(".py"): + raise ScriptedGameplayAuditError("audit path is not authored map Python: {}".format(value)) + maps_root = (root / "maps").resolve() + native = (root / value).resolve() + if native == maps_root or maps_root not in native.parents: + raise ScriptedGameplayAuditError("audit path escapes authored maps: {}".format(value)) + return value + + +def _literal_metric(call: ast.Call, path: str) -> str: + if not call.args or not isinstance(call.args[0], ast.Constant) or not isinstance( + call.args[0].value, str + ): + raise ScriptedGameplayAuditError( + "{}:{} uses a dynamic metric identity".format(path, call.lineno) + ) + return call.args[0].value + + +def _source_paths(root: Path) -> Iterator[Path]: + maps_root = root / "maps" + for source in sorted(maps_root.rglob("*.py")): + relative = source.relative_to(maps_root) + if relative.parts[:2] == ("python", "tests"): + continue + yield source + + +def _children(node: ast.AST) -> Iterator[tuple[str, ast.AST]]: + for field, value in ast.iter_fields(node): + if isinstance(value, ast.AST): + yield field, value + elif isinstance(value, list): + for index, item in enumerate(value): + if isinstance(item, ast.AST): + yield "{}[{}]".format(field, index), item + + +def _context_sha256(node: ast.AST) -> str: + serialized = ast.dump(node, annotate_fields=True, include_attributes=False) + return hashlib.sha256(serialized.encode("utf-8")).hexdigest() + + +def _static_string(node: ast.AST) -> str | None: + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return node.value + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): + left = _static_string(node.left) + right = _static_string(node.right) + if left is not None and right is not None: + return left + right + if isinstance(node, ast.JoinedStr): + parts: list[str] = [] + for value in node.values: + part = _static_string(value) + if part is None: + return None + parts.append(part) + return "".join(parts) + return None + + +def _sensitive_receiver(node: ast.AST, aliases: set[str]) -> bool: + if isinstance(node, ast.Name): + return node.id in aliases + if isinstance(node, ast.Attribute): + return _sensitive_receiver(node.value, aliases) + if isinstance(node, ast.Subscript): + return _sensitive_receiver(node.value, aliases) + if isinstance(node, (ast.Tuple, ast.List, ast.Set)): + return any(_sensitive_receiver(value, aliases) for value in node.elts) + if isinstance(node, ast.Call): + if isinstance(node.func, ast.Attribute): + return node.func.attr == "Controller" + if isinstance(node.func, ast.Name): + if node.func.id == "type" and node.args: + return _sensitive_receiver(node.args[0], aliases) + return node.func.id in {"FindPlayer", "GetFirst", "Guild", "WhoIsActivator"} + if isinstance(node, ast.IfExp): + return _sensitive_receiver(node.body, aliases) or _sensitive_receiver( + node.orelse, aliases + ) + if isinstance(node, ast.BoolOp): + return any(_sensitive_receiver(value, aliases) for value in node.values) + return False + + +def _contains_sensitive_receiver(node: ast.AST, aliases: set[str]) -> bool: + return any(_sensitive_receiver(child, aliases) for child in ast.walk(node)) + + +def _loaded_names(node: ast.AST) -> set[str]: + return { + child.id + for child in ast.walk(node) + if isinstance(child, ast.Name) and isinstance(child.ctx, ast.Load) + } + + +def _interactive_console_receiver(node: ast.AST, code_aliases: set[str]) -> bool: + return ( + isinstance(node, ast.Attribute) + and node.attr == "InteractiveConsole" + and isinstance(node.value, ast.Name) + and node.value.id in code_aliases + ) + + +def _propagate_alias_target( + target: ast.AST, + value: ast.AST, + aliases: set[str], +) -> bool: + if isinstance(target, ast.Name) and _sensitive_receiver(value, aliases): + if target.id not in aliases: + aliases.add(target.id) + return True + return False + if isinstance(target, (ast.Tuple, ast.List)) and isinstance(value, (ast.Tuple, ast.List)): + changed = False + for target_item, value_item in zip(target.elts, value.elts): + changed = _propagate_alias_target(target_item, value_item, aliases) or changed + return changed + return False + + +def _target_value_pairs(target: ast.AST, value: ast.AST) -> Iterator[tuple[str, ast.AST]]: + """Yield statically paired assignment names for conservative rebinding checks.""" + + if isinstance(target, ast.Name): + yield target.id, value + elif isinstance(target, (ast.Tuple, ast.List)) and isinstance( + value, (ast.Tuple, ast.List) + ): + for target_item, value_item in zip(target.elts, value.elts): + yield from _target_value_pairs(target_item, value_item) + + +def _discover_source(root: Path, source: Path) -> tuple[list[dict[str, str]], list[dict[str, str]]]: + relative = source.relative_to(root).as_posix() + try: + tree = ast.parse(source.read_text(encoding="utf-8"), filename=relative) + except (OSError, UnicodeError, SyntaxError) as error: + raise ScriptedGameplayAuditError( + "cannot inspect authored Python {}: {}".format(relative, error) + ) from error + + metrics: list[dict[str, str]] = [] + logs: list[dict[str, str]] = [] + direct_metric_attributes: set[int] = set() + direct_log_references: set[int] = set() + direct_reflection_references: set[int] = set() + sensitive_aliases = set(SENSITIVE_RECEIVER_NAMES) + reflection_aliases = {"getattr", "vars"} + logger_aliases = {"Logger"} + print_aliases = {"print"} + atrinik_aliases = {"Atrinik"} + builtins_aliases = {"__builtins__", "builtins"} + code_aliases = {"code"} + atrinik_wildcard = False + assignments: list[tuple[ast.AST, ast.AST, int]] = [] + ambiguous_sensitive_aliases: set[str] = set() + for imported in ast.walk(tree): + if isinstance(imported, ast.Import): + for name in imported.names: + if name.name == "builtins": + builtins_aliases.add(name.asname or name.name) + if name.name == "Atrinik": + atrinik_aliases.add(name.asname or name.name) + if name.name == "code": + code_aliases.add(name.asname or name.name) + if isinstance(imported, ast.ImportFrom) and imported.module == "Atrinik": + for name in imported.names: + if name.name == "*": + atrinik_wildcard = True + if name.name == "Logger": + logger_aliases.add(name.asname or name.name) + if name.name == "print": + raise ScriptedGameplayAuditError( + "{}:{} aliases a reserved audit callable".format( + relative, imported.lineno + ) + ) + if name.name == "Eval": + raise ScriptedGameplayAuditError( + "{}:{} aliases a reserved execution callable".format( + relative, imported.lineno + ) + ) + if isinstance(imported, ast.ImportFrom) and imported.module == "code": + if any(name.name in {"*", "InteractiveConsole"} for name in imported.names): + raise ScriptedGameplayAuditError( + "{}:{} aliases a reserved execution facility".format( + relative, imported.lineno + ) + ) + if isinstance(imported, ast.ImportFrom) and imported.module == "builtins": + if any( + name.name in {"getattr", "print", "vars"} | DYNAMIC_EXECUTION_NAMES + | NAMESPACE_REFLECTION_NAMES + for name in imported.names + ): + raise ScriptedGameplayAuditError( + "{}:{} aliases a reserved reflective callable".format( + relative, imported.lineno + ) + ) + if isinstance(imported, ast.ImportFrom) and imported.module == "operator": + if any( + name.name in {"*", "attrgetter", "methodcaller"} + for name in imported.names + ): + raise ScriptedGameplayAuditError( + "{}:{} aliases a reserved reflective callable".format( + relative, imported.lineno + ) + ) + if isinstance(imported, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)): + arguments = ( + [*imported.args.posonlyargs, *imported.args.args, *imported.args.kwonlyargs] + + ([imported.args.vararg] if imported.args.vararg is not None else []) + + ([imported.args.kwarg] if imported.args.kwarg is not None else []) + ) + if any( + argument.arg in logger_aliases | print_aliases + for argument in arguments + ): + raise ScriptedGameplayAuditError( + "{}:{} shadows a reserved audit callable".format(relative, imported.lineno) + ) + if isinstance(imported, ast.Assign): + assignments.extend( + (target, imported.value, imported.lineno) for target in imported.targets + ) + elif isinstance(imported, ast.AnnAssign) and imported.value is not None: + assignments.append((imported.target, imported.value, imported.lineno)) + elif isinstance(imported, ast.NamedExpr): + assignments.append((imported.target, imported.value, imported.lineno)) + if atrinik_wildcard: + for binding in ast.walk(tree): + if ( + isinstance(binding, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) + and binding.name == "Eval" + ): + raise ScriptedGameplayAuditError( + "{}:{} shadows the wildcard Atrinik.Eval binding".format( + relative, binding.lineno + ) + ) + if isinstance(binding, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)): + arguments = ( + [ + *binding.args.posonlyargs, + *binding.args.args, + *binding.args.kwonlyargs, + ] + + ([binding.args.vararg] if binding.args.vararg is not None else []) + + ([binding.args.kwarg] if binding.args.kwarg is not None else []) + ) + if any(argument.arg == "Eval" for argument in arguments): + raise ScriptedGameplayAuditError( + "{}:{} shadows the wildcard Atrinik.Eval binding".format( + relative, binding.lineno + ) + ) + if ( + isinstance(binding, ast.Name) + and isinstance(binding.ctx, (ast.Store, ast.Del)) + and binding.id == "Eval" + ): + raise ScriptedGameplayAuditError( + "{}:{} shadows the wildcard Atrinik.Eval binding".format( + relative, binding.lineno + ) + ) + if isinstance(binding, (ast.Import, ast.ImportFrom)) and any( + (name.asname or name.name.split(".")[0]) == "Eval" + for name in binding.names + if name.name != "*" + ): + raise ScriptedGameplayAuditError( + "{}:{} shadows the wildcard Atrinik.Eval binding".format( + relative, binding.lineno + ) + ) + if ( + isinstance(binding, ast.ImportFrom) + and binding.module != "Atrinik" + and any(name.name == "*" for name in binding.names) + ): + raise ScriptedGameplayAuditError( + "{}:{} ambiguously overwrites wildcard Atrinik bindings".format( + relative, binding.lineno + ) + ) + if ( + isinstance(binding, (ast.ExceptHandler, ast.MatchAs, ast.MatchStar)) + and binding.name == "Eval" + ) or ( + isinstance(binding, ast.MatchMapping) and binding.rest == "Eval" + ): + raise ScriptedGameplayAuditError( + "{}:{} shadows the wildcard Atrinik.Eval binding".format( + relative, binding.lineno + ) + ) + changed = True + while changed: + changed = False + for target, value, lineno in assignments: + for _name, assigned_value in _target_value_pairs(target, value): + if isinstance(assigned_value, ast.Name): + if ( + assigned_value.id in builtins_aliases + and _name not in builtins_aliases + ): + builtins_aliases.add(_name) + changed = True + if ( + assigned_value.id in atrinik_aliases + and _name not in atrinik_aliases + ): + atrinik_aliases.add(_name) + changed = True + if assigned_value.id in code_aliases and _name not in code_aliases: + code_aliases.add(_name) + changed = True + if ( + isinstance(assigned_value, ast.Attribute) + and isinstance(assigned_value.value, ast.Name) + and assigned_value.value.id in builtins_aliases + and assigned_value.attr + in {"getattr", "vars"} + | DYNAMIC_EXECUTION_NAMES + | NAMESPACE_REFLECTION_NAMES + ): + raise ScriptedGameplayAuditError( + "{}:{} aliases a reserved reflective callable".format( + relative, lineno + ) + ) + if ( + isinstance(assigned_value, ast.Attribute) + and assigned_value.attr == "InteractiveConsole" + and isinstance(assigned_value.value, ast.Name) + and assigned_value.value.id in code_aliases + ): + raise ScriptedGameplayAuditError( + "{}:{} aliases a reserved execution facility".format( + relative, lineno + ) + ) + if isinstance(target, ast.Name) and target.id in {"Logger", "print"}: + raise ScriptedGameplayAuditError( + "{}:{} shadows a reserved audit callable".format(relative, lineno) + ) + if ( + isinstance(target, ast.Name) + and target.id == "guild" + and not _sensitive_receiver(value, sensitive_aliases) + ): + raise ScriptedGameplayAuditError( + "{}:{} ambiguously rebinds the guild audit receiver".format( + relative, lineno + ) + ) + changed = _propagate_alias_target(target, value, sensitive_aliases) or changed + if ( + isinstance(value, ast.Attribute) + and isinstance(value.value, ast.Name) + and value.value.id in builtins_aliases + and value.attr + in {"getattr", "vars"} + | DYNAMIC_EXECUTION_NAMES + | NAMESPACE_REFLECTION_NAMES + ): + raise ScriptedGameplayAuditError( + "{}:{} aliases a reserved reflective callable".format(relative, lineno) + ) + if ( + isinstance(value, ast.Attribute) + and value.attr in {"get", "__getitem__"} + and isinstance(value.value, ast.Call) + and isinstance(value.value.func, ast.Name) + and value.value.func.id in {"globals", "locals"} + ): + raise ScriptedGameplayAuditError( + "{}:{} aliases a reflective namespace lookup".format(relative, lineno) + ) + if ( + isinstance(value, ast.Call) + and isinstance(value.func, ast.Name) + and value.func.id in {"globals", "locals"} + ) or ( + isinstance(value, ast.Name) + and value.id in {"globals", "locals"} + ): + raise ScriptedGameplayAuditError( + "{}:{} aliases a reflective namespace".format(relative, lineno) + ) + if ( + isinstance(value, ast.Attribute) + and isinstance(value.value, ast.Name) + and ( + value.value.id in atrinik_aliases + and value.attr == "print" + or value.value.id in builtins_aliases + and value.attr + in {"print"} | DYNAMIC_EXECUTION_NAMES | NAMESPACE_REFLECTION_NAMES + ) + ): + raise ScriptedGameplayAuditError( + "{}:{} aliases a reserved audit callable".format(relative, lineno) + ) + if not isinstance(target, ast.Name): + continue + if isinstance(value, ast.Name): + if value.id in builtins_aliases and target.id not in builtins_aliases: + builtins_aliases.add(target.id) + changed = True + if value.id in atrinik_aliases and target.id not in atrinik_aliases: + atrinik_aliases.add(target.id) + changed = True + if value.id in code_aliases and target.id not in code_aliases: + code_aliases.add(target.id) + changed = True + if value.id in reflection_aliases and target.id not in reflection_aliases: + reflection_aliases.add(target.id) + changed = True + if value.id in logger_aliases and target.id not in logger_aliases: + logger_aliases.add(target.id) + changed = True + if value.id in print_aliases and target.id not in print_aliases: + print_aliases.add(target.id) + changed = True + if value.id in DYNAMIC_EXECUTION_NAMES: + raise ScriptedGameplayAuditError( + "{}:{} aliases a reserved reflective callable".format(relative, lineno) + ) + if target.id in print_aliases and not ( + isinstance(value, ast.Name) and value.id in print_aliases + ): + raise ScriptedGameplayAuditError( + "{}:{} ambiguously rebinds a print alias".format(relative, lineno) + ) + if ( + isinstance(value, ast.Attribute) + and value.attr == "Logger" + and isinstance(value.value, ast.Name) + and value.value.id == "Atrinik" + and target.id not in logger_aliases + ): + logger_aliases.add(target.id) + changed = True + + for target, value, lineno in assignments: + for name, assigned_value in _target_value_pairs(target, value): + if name in sensitive_aliases and not _sensitive_receiver( + assigned_value, sensitive_aliases + ): + ambiguous_sensitive_aliases.add(name) + + def visit( + node: ast.AST, + ast_path: str, + scopes: tuple[str, ...], + context_sha256: str, + ) -> None: + next_scopes = scopes + next_context_sha256 = context_sha256 + if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): + next_scopes = (*scopes, node.name) + next_context_sha256 = _context_sha256(node) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and ( + (*scopes, node.name) == ("Guild", "log_add") + ): + logs.append( + { + "path": relative, + "ast_path": ast_path, + "scope": "Guild.log_add", + "context_sha256": next_context_sha256, + "facility": "Guild.log_add sink", + } + ) + if isinstance(node, ast.Call): + scope = ".".join(scopes) or "" + if isinstance(node.func, ast.Attribute) and node.func.attr in METRIC_METHODS: + if _loaded_names(node.func.value) & ambiguous_sensitive_aliases: + raise ScriptedGameplayAuditError( + "{}:{} uses an ambiguously rebound telemetry receiver".format( + relative, node.lineno + ) + ) + if not _sensitive_receiver(node.func.value, sensitive_aliases): + raise ScriptedGameplayAuditError( + "{}:{} uses a reserved metric method on an unreviewed receiver".format( + relative, node.lineno + ) + ) + direct_metric_attributes.add(id(node.func)) + metrics.append( + { + "path": relative, + "ast_path": ast_path, + "scope": scope, + "context_sha256": context_sha256, + "method": node.func.attr, + "metric": _literal_metric(node, relative), + } + ) + facility = None + if isinstance(node.func, ast.Name) and node.func.id in logger_aliases: + facility = "Logger" if node.func.id == "Logger" else "Atrinik.Logger" + direct_log_references.add(id(node.func)) + elif isinstance(node.func, ast.Name) and node.func.id in print_aliases: + facility = "Python.print" + direct_log_references.add(id(node.func)) + elif ( + isinstance(node.func, ast.Attribute) + and node.func.attr == "Logger" + and isinstance(node.func.value, ast.Name) + and node.func.value.id in atrinik_aliases + ): + facility = "Atrinik.Logger" + direct_log_references.add(id(node.func)) + elif ( + isinstance(node.func, ast.Attribute) + and node.func.attr == "print" + and isinstance(node.func.value, ast.Name) + and node.func.value.id in builtins_aliases | atrinik_aliases + ): + facility = "Python.print" + direct_log_references.add(id(node.func)) + elif ( + isinstance(node.func, ast.Attribute) + and node.func.attr == "log_add" + and _sensitive_receiver(node.func.value, sensitive_aliases) + ): + if _loaded_names(node.func.value) & ambiguous_sensitive_aliases: + raise ScriptedGameplayAuditError( + "{}:{} uses an ambiguously rebound telemetry receiver".format( + relative, node.lineno + ) + ) + facility = "Guild.log_add" + direct_log_references.add(id(node.func)) + if facility is not None: + logs.append( + { + "path": relative, + "ast_path": ast_path, + "scope": scope, + "context_sha256": context_sha256, + "facility": facility, + } + ) + if isinstance(node.func, ast.Name) and node.func.id in {"eval", "exec"}: + logs.append( + { + "path": relative, + "ast_path": ast_path, + "scope": scope, + "context_sha256": context_sha256, + "facility": "Python.{}".format(node.func.id), + } + ) + direct_log_references.add(id(node.func)) + if ( + atrinik_wildcard + and isinstance(node.func, ast.Name) + and node.func.id == "Eval" + ): + logs.append( + { + "path": relative, + "ast_path": ast_path, + "scope": scope, + "context_sha256": context_sha256, + "facility": "Atrinik.Eval", + } + ) + direct_log_references.add(id(node.func)) + if ( + isinstance(node.func, ast.Attribute) + and isinstance(node.func.value, ast.Name) + and node.func.value.id in builtins_aliases + and node.func.attr in {"eval", "exec"} + ): + logs.append( + { + "path": relative, + "ast_path": ast_path, + "scope": scope, + "context_sha256": context_sha256, + "facility": "Python.{}".format(node.func.attr), + } + ) + direct_log_references.add(id(node.func)) + if ( + isinstance(node.func, ast.Attribute) + and node.func.attr == "push" + and isinstance(node.func.value, ast.Attribute) + and node.func.value.attr == "InteractiveConsole" + and isinstance(node.func.value.value, ast.Name) + and node.func.value.value.id in code_aliases + ): + logs.append( + { + "path": relative, + "ast_path": ast_path, + "scope": scope, + "context_sha256": context_sha256, + "facility": "Python.InteractiveConsole.push", + } + ) + direct_log_references.add(id(node.func)) + if ( + isinstance(node.func, ast.Name) + and node.func.id in {"getattr", "vars"} + ): + direct_reflection_references.add(id(node.func)) + for edge, child in _children(node): + visit( + child, + "{}.{}".format(ast_path, edge) if ast_path else edge, + next_scopes, + next_context_sha256, + ) + + visit(tree, "", (), _context_sha256(tree)) + for node in ast.walk(tree): + if isinstance(node, ast.Attribute) and node.attr in METRIC_METHODS: + if id(node) not in direct_metric_attributes: + raise ScriptedGameplayAuditError( + "{}:{} uses an indirect metric method reference".format(relative, node.lineno) + ) + if ( + ( + isinstance(node, ast.Name) + and isinstance(node.ctx, ast.Load) + and node.id in logger_aliases + ) + or ( + isinstance(node, ast.Attribute) + and ( + ( + node.attr == "Logger" + ) + or ( + node.attr == "log_add" + ) + or ( + node.attr == "print" + and isinstance(node.value, ast.Name) + and node.value.id in builtins_aliases | atrinik_aliases + ) + or ( + node.attr in {"eval", "exec"} + and isinstance(node.value, ast.Name) + and node.value.id in builtins_aliases + ) + or ( + node.attr == "Eval" + and isinstance(node.value, ast.Name) + and node.value.id in atrinik_aliases + ) + or ( + node.attr == "push" + and isinstance(node.value, ast.Attribute) + and node.value.attr == "InteractiveConsole" + and isinstance(node.value.value, ast.Name) + and node.value.value.id in code_aliases + ) + ) + ) + ) and id(node) not in direct_log_references: + raise ScriptedGameplayAuditError( + "{}:{} uses an indirect audit-log reference".format(relative, node.lineno) + ) + if ( + isinstance(node, ast.Name) + and isinstance(node.ctx, ast.Load) + and node.id in {"getattr", "vars", "print"} | DYNAMIC_EXECUTION_NAMES + and id(node) not in direct_reflection_references + and id(node) not in direct_log_references + ): + raise ScriptedGameplayAuditError( + "{}:{} aliases a reserved reflective callable".format(relative, node.lineno) + ) + if ( + atrinik_wildcard + and isinstance(node, ast.Name) + and isinstance(node.ctx, ast.Load) + and node.id == "Eval" + and id(node) not in direct_log_references + ): + raise ScriptedGameplayAuditError( + "{}:{} aliases a reserved execution callable".format(relative, node.lineno) + ) + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id in reflection_aliases + and any(isinstance(argument, ast.Starred) for argument in node.args) + ): + raise ScriptedGameplayAuditError( + "{}:{} uses reflective telemetry access".format(relative, node.lineno) + ) + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id in reflection_aliases + and node.func.id != "vars" + and len(node.args) >= 2 + ): + reflected_name = _static_string(node.args[1]) + if reflected_name in ( + SENSITIVE_REFLECTIVE_NAMES + | DYNAMIC_EXECUTION_NAMES + | NAMESPACE_REFLECTION_NAMES + ) or ( + reflected_name == "InteractiveConsole" + and isinstance(node.args[0], ast.Name) + and node.args[0].id in code_aliases + ) or ( + reflected_name == "push" + and _interactive_console_receiver(node.args[0], code_aliases) + ) or ( + reflected_name is None + and ( + _sensitive_receiver(node.args[0], sensitive_aliases) + or ( + isinstance(node.args[0], ast.Name) + and node.args[0].id in builtins_aliases + ) + or ( + isinstance(node.args[0], ast.Name) + and node.args[0].id in code_aliases + ) + or _interactive_console_receiver(node.args[0], code_aliases) + ) + ): + raise ScriptedGameplayAuditError( + "{}:{} uses reflective telemetry access".format(relative, node.lineno) + ) + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id in reflection_aliases + and node.func.id != "getattr" + and node.args + and ( + _sensitive_receiver(node.args[0], sensitive_aliases) + or ( + isinstance(node.args[0], ast.Name) + and node.args[0].id in builtins_aliases + ) + or ( + isinstance(node.args[0], ast.Name) + and node.args[0].id in code_aliases + ) + ) + ): + raise ScriptedGameplayAuditError( + "{}:{} uses reflective telemetry access".format(relative, node.lineno) + ) + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and isinstance(node.func.value, ast.Name) + and node.func.value.id in builtins_aliases + and node.func.attr + in {"compile", "getattr", "vars", "__import__"} + | NAMESPACE_REFLECTION_NAMES + ): + raise ScriptedGameplayAuditError( + "{}:{} uses qualified reflective access".format(relative, node.lineno) + ) + if ( + isinstance(node, ast.Attribute) + and node.attr == "__getattribute__" + ): + raise ScriptedGameplayAuditError( + "{}:{} uses reflective telemetry access".format(relative, node.lineno) + ) + if ( + isinstance(node, ast.Attribute) + and node.attr == "__dict__" + and ( + _contains_sensitive_receiver(node.value, sensitive_aliases) + or ( + isinstance(node.value, ast.Name) + and node.value.id in builtins_aliases | code_aliases + ) + or _interactive_console_receiver(node.value, code_aliases) + ) + ): + raise ScriptedGameplayAuditError( + "{}:{} uses reflective telemetry access".format(relative, node.lineno) + ) + if isinstance(node, ast.Attribute) and node.attr in {"attrgetter", "methodcaller"}: + raise ScriptedGameplayAuditError( + "{}:{} uses reflective attribute construction".format(relative, node.lineno) + ) + if ( + isinstance(node, ast.Subscript) + and isinstance(node.value, ast.Attribute) + and node.value.attr == "__dict__" + and _static_string(node.slice) + in SENSITIVE_REFLECTIVE_NAMES + | DYNAMIC_EXECUTION_NAMES + | {"attrgetter", "methodcaller"} + ): + raise ScriptedGameplayAuditError( + "{}:{} uses reflective telemetry access".format(relative, node.lineno) + ) + if ( + isinstance(node, ast.Subscript) + and ( + ( + isinstance(node.value, ast.Name) + and node.value.id == "__builtins__" + ) + or ( + isinstance(node.value, ast.Call) + and isinstance(node.value.func, ast.Name) + and node.value.func.id in {"globals", "locals"} + ) + ) + ): + raise ScriptedGameplayAuditError( + "{}:{} uses reflective namespace access".format(relative, node.lineno) + ) + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr in {"get", "__getitem__"} + and ( + ( + isinstance(node.func.value, ast.Call) + and isinstance(node.func.value.func, ast.Name) + and node.func.value.func.id in {"globals", "locals"} + ) + or ( + isinstance(node.func.value, ast.Name) + and node.func.value.id in builtins_aliases + ) + ) + ): + raise ScriptedGameplayAuditError( + "{}:{} uses reflective namespace access".format(relative, node.lineno) + ) + return metrics, logs + + +def discover_sites(root: Path) -> tuple[list[dict[str, str]], list[dict[str, str]]]: + """Return every exact authored metric and audit-log call occurrence.""" + + metrics: list[dict[str, str]] = [] + logs: list[dict[str, str]] = [] + for source in _source_paths(root): + source_metrics, source_logs = _discover_source(root, source) + metrics.extend(source_metrics) + logs.extend(source_logs) + metrics.sort(key=lambda row: (row["path"], row["ast_path"])) + logs.sort(key=lambda row: (row["path"], row["ast_path"])) + return metrics, logs + + +def _require_text(row: dict[str, Any], field: str, context: str) -> str: + value = row.get(field) + if not isinstance(value, str) or not value.strip(): + raise ScriptedGameplayAuditError("{} requires non-empty {}".format(context, field)) + return value + + +def _validate_identity(value: object, pattern: re.Pattern[str], context: str) -> str: + if not isinstance(value, str) or not value.isascii(): + raise ScriptedGameplayAuditError("{} must be bounded ASCII".format(context)) + if not value or len(value) > IDENTITY_MAX or pattern.fullmatch(value) is None: + raise ScriptedGameplayAuditError( + "{} is invalid or exceeds {} characters".format(context, IDENTITY_MAX) + ) + return value + + +def _validate_common_source( + row: dict[str, Any], root: Path, context: str +) -> tuple[str, str, str]: + source = _safe_relative_path(root, row["path"]) + ast_path = _require_text(row, "ast_path", context) + scope = _require_text(row, "scope", context) + if not (root / source).is_file(): + raise ScriptedGameplayAuditError("{} source is missing".format(context)) + return source, ast_path, scope + + +def load_and_validate(root: Path) -> dict[str, Any]: + """Validate the closed manifest and its exact source inventory.""" + + path = root / MANIFEST_PATH + try: + document = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as error: + raise ScriptedGameplayAuditError( + "cannot load {}: {}".format(MANIFEST_PATH, error) + ) from error + if not isinstance(document, dict) or set(document) != { + "schema_version", + "source_contexts", + "metric_sites", + "audit_like_sites", + }: + raise ScriptedGameplayAuditError("scripted gameplay audit must be a closed v1 object") + if document["schema_version"] != 1: + raise ScriptedGameplayAuditError("unsupported scripted gameplay audit schema") + + expected_contexts: dict[tuple[str, str], str] = {} + context_rows = document["source_contexts"] + if not isinstance(context_rows, list) or not context_rows: + raise ScriptedGameplayAuditError("source_contexts must be a non-empty array") + context_order: list[tuple[str, str]] = [] + for index, row in enumerate(context_rows): + context = "source_contexts[{}]".format(index) + if not isinstance(row, dict) or set(row) != {"path", "scope", "context_sha256"}: + raise ScriptedGameplayAuditError("{} has an unknown or missing field".format(context)) + source = _safe_relative_path(root, row["path"]) + scope = _require_text(row, "scope", context) + context_sha256 = row["context_sha256"] + if ( + not isinstance(context_sha256, str) + or re.fullmatch(r"[0-9a-f]{64}", context_sha256) is None + ): + raise ScriptedGameplayAuditError("{} has an invalid context_sha256".format(context)) + if not (root / source).is_file(): + raise ScriptedGameplayAuditError("{} source is missing".format(context)) + key = (source, scope) + context_order.append(key) + expected_contexts[key] = context_sha256 + if context_order != sorted(context_order) or len(context_order) != len(set(context_order)): + raise ScriptedGameplayAuditError("source_contexts must be sorted and unique") + + expected_metrics: list[dict[str, str]] = [] + rows = document["metric_sites"] + if not isinstance(rows, list) or not rows: + raise ScriptedGameplayAuditError("metric_sites must be a non-empty array") + required = { + "path", + "ast_path", + "scope", + "method", + "metric", + "classification", + "proposed_journal_reason", + "event_rate", + "rationale", + } + ordering: list[tuple[str, str]] = [] + for index, row in enumerate(rows): + context = "metric_sites[{}]".format(index) + if not isinstance(row, dict) or set(row) != required: + raise ScriptedGameplayAuditError("{} has an unknown or missing field".format(context)) + source, ast_path, scope = _validate_common_source(row, root, context) + context_sha256 = expected_contexts.get((source, scope)) + if context_sha256 is None: + raise ScriptedGameplayAuditError("{} has no reviewed source context".format(context)) + method = row["method"] + if method not in METRIC_METHODS: + raise ScriptedGameplayAuditError("{} has an unknown metric method".format(context)) + metric = _validate_identity(row["metric"], METRIC_ID, "{} metric".format(context)) + classification = row["classification"] + if classification not in CLASSIFICATIONS: + raise ScriptedGameplayAuditError("{} has an unknown classification".format(context)) + reason = row["proposed_journal_reason"] + if classification == "gameplay-journal": + _validate_identity(reason, JOURNAL_REASON, "{} proposed reason".format(context)) + elif reason is not None: + raise ScriptedGameplayAuditError( + "{} must not propose a journal reason for {}".format(context, classification) + ) + _require_text(row, "event_rate", context) + _require_text(row, "rationale", context) + expected_metrics.append( + { + "path": source, + "ast_path": ast_path, + "scope": scope, + "context_sha256": context_sha256, + "method": method, + "metric": metric, + } + ) + ordering.append((source, ast_path)) + if ordering != sorted(ordering) or len(ordering) != len(set(ordering)): + raise ScriptedGameplayAuditError("metric_sites must be sorted and occurrence-unique") + + expected_logs: list[dict[str, str]] = [] + audit_rows = document["audit_like_sites"] + if not isinstance(audit_rows, list) or not audit_rows: + raise ScriptedGameplayAuditError("audit_like_sites must be a non-empty array") + audit_required = { + "path", + "ast_path", + "scope", + "facility", + "classification", + "event_rate", + "rationale", + } + audit_order: list[tuple[str, str]] = [] + for index, row in enumerate(audit_rows): + context = "audit_like_sites[{}]".format(index) + if not isinstance(row, dict) or set(row) != audit_required: + raise ScriptedGameplayAuditError("{} has an unknown or missing field".format(context)) + source, ast_path, scope = _validate_common_source(row, root, context) + context_sha256 = expected_contexts.get((source, scope)) + if context_sha256 is None: + raise ScriptedGameplayAuditError("{} has no reviewed source context".format(context)) + facility = _require_text(row, "facility", context) + classification = row["classification"] + if classification not in {"operational/security-log", "not-recorded"}: + raise ScriptedGameplayAuditError( + "{} audit facility must remain operational or not recorded".format(context) + ) + _require_text(row, "event_rate", context) + _require_text(row, "rationale", context) + expected_logs.append( + { + "path": source, + "ast_path": ast_path, + "scope": scope, + "context_sha256": context_sha256, + "facility": facility, + } + ) + audit_order.append((source, ast_path)) + if audit_order != sorted(audit_order) or len(audit_order) != len(set(audit_order)): + raise ScriptedGameplayAuditError("audit_like_sites must be sorted and occurrence-unique") + + actual_metrics, actual_logs = discover_sites(root) + actual_contexts = { + (row["path"], row["scope"]): row["context_sha256"] + for row in (*actual_metrics, *actual_logs) + } + if actual_contexts != expected_contexts: + raise ScriptedGameplayAuditError( + "source-context inventory differs: expected={} actual={}".format( + expected_contexts, actual_contexts + ) + ) + if actual_metrics != expected_metrics: + raise ScriptedGameplayAuditError( + "metric-site inventory differs: expected={} actual={}".format( + expected_metrics, actual_metrics + ) + ) + if actual_logs != expected_logs: + raise ScriptedGameplayAuditError( + "audit-like-site inventory differs: expected={} actual={}".format( + expected_logs, actual_logs + ) + ) + return { + "metric_sites": len(actual_metrics), + "metric_identities": len({row["metric"] for row in actual_metrics}), + "audit_like_sites": len(actual_logs), + } diff --git a/tools/tests/test_pr_metadata.py b/tools/tests/test_pr_metadata.py index 85441e65e..ba53e1231 100644 --- a/tools/tests/test_pr_metadata.py +++ b/tools/tests/test_pr_metadata.py @@ -58,6 +58,7 @@ def test_workflow_uses_trusted_policy_and_aggregate_runs_tests(self) -> None: self.assertIn(".github/scripts/check_pr_metadata.py", policy_workflow) self.assertIn("python3 tools/validate.py", check_workflow) self.assertIn("tools.tests.test_pr_metadata", aggregate) + self.assertIn("tools.tests.test_scripted_gameplay_audit", aggregate) if __name__ == "__main__": diff --git a/tools/tests/test_scripted_gameplay_audit.py b/tools/tests/test_scripted_gameplay_audit.py new file mode 100644 index 000000000..ddd828d74 --- /dev/null +++ b/tools/tests/test_scripted_gameplay_audit.py @@ -0,0 +1,854 @@ +from __future__ import annotations + +import json +from pathlib import Path +import tempfile +import unittest + +from tools.scripted_gameplay_audit import ( + MANIFEST_PATH, + ScriptedGameplayAuditError, + discover_sites, + load_and_validate, +) + + +ROOT = Path(__file__).resolve().parents[2] + + +class ScriptedGameplayAuditTests(unittest.TestCase): + def _fixture( + self, + root: Path, + source_text: str, + source_path: str = "maps/python/feature.py", + ) -> Path: + source = root / source_path + source.parent.mkdir(parents=True, exist_ok=True) + source.write_text(source_text, encoding="utf-8") + metrics, logs = discover_sites(root) + sites = [*metrics, *logs] + document = { + "schema_version": 1, + "source_contexts": [ + { + "path": path, + "scope": scope, + "context_sha256": context_sha256, + } + for (path, scope), context_sha256 in sorted({ + (row["path"], row["scope"]): row["context_sha256"] + for row in sites + }.items()) + ], + "metric_sites": [ + { + **{key: value for key, value in row.items() if key != "context_sha256"}, + "classification": "gameplay-journal", + "proposed_journal_reason": "fixture.event", + "event_rate": "bounded", + "rationale": "fixture", + } + for row in metrics + ], + "audit_like_sites": [ + { + **{key: value for key, value in row.items() if key != "context_sha256"}, + "classification": "operational/security-log", + "event_rate": "bounded", + "rationale": "fixture", + } + for row in logs + ], + } + target = root / MANIFEST_PATH + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(json.dumps(document), encoding="utf-8") + return target + + def test_repository_inventory_is_exact_and_reviewed(self): + report = load_and_validate(ROOT) + self.assertEqual(26, report["metric_sites"]) + self.assertEqual(21, report["metric_identities"]) + self.assertEqual(21, report["audit_like_sites"]) + + def test_unreviewed_metric_site_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + target = self._fixture( + root, + 'player.MetricAdd("economy.new_action")\nLogger("fixture")\n', + ) + document = json.loads(target.read_text(encoding="utf-8")) + document["metric_sites"][0]["metric"] = "economy.other_action" + target.write_text(json.dumps(document), encoding="utf-8") + with self.assertRaisesRegex(ScriptedGameplayAuditError, "inventory differs"): + load_and_validate(root) + + def test_dynamic_metric_identity_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text("player.MetricAdd(metric_name)\n", encoding="utf-8") + with self.assertRaisesRegex(ScriptedGameplayAuditError, "dynamic metric identity"): + discover_sites(root) + + def test_metric_alias_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'emit = player.MetricAdd\nemit("economy.new_action")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "indirect metric method"): + discover_sites(root) + + def test_metric_getattr_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'getattr(player, "MetricAdd")("economy.new_action")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reflective telemetry"): + discover_sites(root) + + def test_audit_log_alias_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'emit = guild.log_add\nemit("player text")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "indirect audit-log"): + discover_sites(root) + + def test_audit_log_getattr_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'getattr(guild, "log_add")("player text")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reflective telemetry"): + discover_sites(root) + + def test_computed_metric_getattr_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'getattr(player, "Metric" + "Add")("economy.hidden")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reflective telemetry"): + discover_sites(root) + + def test_dynamic_aliased_metric_getattr_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'subject = player\ngetattr(subject, f"{kind}Add")("economy.hidden")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reflective telemetry"): + discover_sites(root) + + def test_metric_dict_reflection_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'player.__dict__["Metric" + "Add"]("economy.hidden")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reflective telemetry"): + discover_sites(root) + + def test_metric_class_dict_reflection_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'player.__class__.__dict__["Metric" + "Add"]("economy.hidden")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reflective telemetry"): + discover_sites(root) + + def test_aliased_getattr_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + "lookup = getattr\nlookup(player, method)(\"economy.hidden\")\n", + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reserved reflective"): + discover_sites(root) + + def test_destructured_receiver_alias_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + "subject, = (player,)\ngetattr(subject, method)(\"economy.hidden\")\n", + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reflective telemetry"): + discover_sites(root) + + def test_getattribute_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + "player.__getattribute__(method)(\"economy.hidden\")\n", + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reflective telemetry"): + discover_sites(root) + + def test_aliased_vars_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + "namespace = vars\nnamespace(player)[method](\"economy.hidden\")\n", + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reserved reflective"): + discover_sites(root) + + def test_imported_logger_alias_is_discovered(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'from Atrinik import Logger as emit\nemit("CHAT", player_text)\n', + encoding="utf-8", + ) + _, logs = discover_sites(root) + self.assertEqual("Atrinik.Logger", logs[0]["facility"]) + + def test_innocent_spelling_is_ignored(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text('label = "Logger"\n', encoding="utf-8") + self.assertEqual(([], []), discover_sites(root)) + + def test_qualified_logger_alias_is_discovered(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'import Atrinik as api\napi.Logger("INFO", "fixture")\n', + encoding="utf-8", + ) + _, logs = discover_sites(root) + self.assertEqual("Atrinik.Logger", logs[0]["facility"]) + + def test_atrinik_print_is_discovered(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'import Atrinik as api\nAtrinik.print("one")\napi.print("two")\n', + encoding="utf-8", + ) + _, logs = discover_sites(root) + self.assertEqual( + ["Python.print", "Python.print"], + [row["facility"] for row in logs], + ) + + def test_qualified_builtin_reflection_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'import builtins as core\ncore.getattr(player, "MetricAdd")("hidden")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "qualified reflective"): + discover_sites(root) + + def test_imported_builtin_alias_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + "from builtins import getattr as lookup\nlookup(player, method)(\"hidden\")\n", + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reserved reflective"): + discover_sites(root) + + def test_qualified_builtin_alias_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + "import builtins\n" + "lookup = builtins.getattr\n" + 'lookup(player, "MetricAdd")("hidden")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reserved reflective"): + discover_sites(root) + + def test_imported_operator_reflection_fails_closed(self): + for callable_name in ("*", "attrgetter", "methodcaller"): + with self.subTest( + callable_name=callable_name + ), tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + ( + "from operator import *\n" + if callable_name == "*" + else "from operator import {} as reflect\n".format( + callable_name + ) + ), + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reserved reflective"): + discover_sites(root) + + def test_atrinik_wildcard_eval_shadowing_fails_closed(self): + for shadow in ( + "Eval = harmless\nEval(payload)\n", + "def run(Eval):\n Eval(payload)\n", + "def Eval(payload):\n return payload\nEval(payload)\n", + "try:\n pass\nexcept Exception as Eval:\n Eval(payload)\n", + "match payload:\n case Eval:\n Eval(payload)\n", + "from harmless import *\nEval(payload)\n", + "Eval(payload)\nfrom harmless import *\n", + ): + with self.subTest( + shadow=shadow + ), tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + "from Atrinik import *\n" + shadow, + encoding="utf-8", + ) + with self.assertRaisesRegex( + ScriptedGameplayAuditError, + "shadows the wildcard|overwrites wildcard Atrinik", + ): + discover_sites(root) + + def test_methodcaller_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'operator.methodcaller("MetricAdd", "hidden")(player)\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "attribute construction"): + discover_sites(root) + + def test_starred_reflection_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'getattr(*(player, "MetricAdd"))("hidden")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reflective telemetry"): + discover_sites(root) + + def test_object_dict_getattribute_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'object.__dict__["__getattribute__"]' + '(player, "MetricAdd")("hidden")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reflective telemetry"): + discover_sites(root) + + def test_dynamic_execution_is_reviewed_and_cannot_be_aliased(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text('eval("player.MetricAdd")\n', encoding="utf-8") + _, logs = discover_sites(root) + self.assertEqual("Python.eval", logs[0]["facility"]) + source.write_text('run = eval\nrun(payload)\n', encoding="utf-8") + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reserved reflective"): + discover_sites(root) + + def test_qualified_dynamic_execution_is_reviewed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'import builtins as runtime\nruntime.eval("payload")\n', + encoding="utf-8", + ) + _, logs = discover_sites(root) + self.assertEqual("Python.eval", logs[0]["facility"]) + + def test_atrinik_wildcard_eval_is_discovered(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + "from Atrinik import *\nEval(player_text)\n", + encoding="utf-8", + ) + _, logs = discover_sites(root) + self.assertEqual("Atrinik.Eval", logs[0]["facility"]) + + def test_dynamic_and_print_attribute_aliases_fail_closed(self): + for source_text in ( + "import Atrinik\nemit = Atrinik.print\n", + "import builtins\nemit = builtins.print\n", + "import builtins\nemit = builtins.eval\n", + "from Atrinik import print as emit\n", + "from Atrinik import Eval as emit\n", + "from code import InteractiveConsole as Runner\n", + "from code import *\n", + "import code\nConsole = code.InteractiveConsole\n", + "import code\nruntime = code\nConsole = runtime.InteractiveConsole\n", + ): + with self.subTest( + source_text=source_text + ), tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text(source_text, encoding="utf-8") + with self.assertRaisesRegex( + ScriptedGameplayAuditError, + "reserved audit|reserved reflective|reserved execution", + ): + discover_sites(root) + + def test_additional_reflection_paths_fail_closed(self): + for source_text in ( + 'import builtins\ngetattr(builtins, "eval")("payload")\n', + 'getattr(player, "__getattribute__")("MetricAdd")\n', + 'import builtins\nvars(builtins)["eval"]("payload")\n', + 'import operator\noperator.__dict__["methodcaller"]("MetricAdd")\n', + "lookup = globals().get\n", + "namespace = globals()\n", + "lookup, = (builtins.getattr,)\n", + "from builtins import globals as namespace\n", + "import builtins\nnamespace = builtins.globals\n", + "import builtins\ngetattr(builtins, method)(payload)\n", + "import builtins\nmodule = builtins\nlookup = module.getattr\n", + 'import code\ngetattr(code, "InteractiveConsole").push(console, payload)\n', + 'import code\ngetattr(code.InteractiveConsole, "push")(console, payload)\n', + 'import code\nvars(code)["InteractiveConsole"].push(console, payload)\n', + 'import code\ncode.__dict__["InteractiveConsole"].push(console, payload)\n', + 'import code\ncode.InteractiveConsole.__dict__["push"](console, payload)\n', + ): + with self.subTest( + source_text=source_text + ), tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text(source_text, encoding="utf-8") + with self.assertRaises(ScriptedGameplayAuditError): + discover_sites(root) + + def test_builtin_module_alias_sites_are_discovered(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + "import builtins\n" + "runtime = builtins\n" + "runtime.eval(payload)\n" + "runtime.print(payload)\n", + encoding="utf-8", + ) + _, logs = discover_sites(root) + self.assertEqual( + ["Python.eval", "Python.print"], + [row["facility"] for row in logs], + ) + + def test_destructured_privileged_module_aliases_are_discovered(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + "import Atrinik, builtins, code\n" + "runtime, = (builtins,)\n" + "api, = (Atrinik,)\n" + "engine, = (code,)\n" + "runtime.eval(payload)\n" + "api.print(payload)\n" + "engine.InteractiveConsole.push(console, payload)\n", + encoding="utf-8", + ) + _, logs = discover_sites(root) + self.assertEqual( + [ + "Python.eval", + "Python.print", + "Python.InteractiveConsole.push", + ], + [row["facility"] for row in logs], + ) + + def test_interactive_console_execution_is_discovered(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + "import code as runtime\n" + "runtime.InteractiveConsole.push(console, player_text)\n", + encoding="utf-8", + ) + _, logs = discover_sites(root) + self.assertEqual( + "Python.InteractiveConsole.push", logs[0]["facility"] + ) + + def test_namespace_get_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text('globals().get("Logger")("hidden")\n', encoding="utf-8") + with self.assertRaisesRegex(ScriptedGameplayAuditError, "namespace access"): + discover_sites(root) + + def test_sensitive_receiver_rebinding_fails_closed(self): + for source_text in ( + 'subject = player\nsubject = cache\nsubject.MetricAdd("cache.hit")\n', + 'def bind():\n subject = player\ndef reuse():\n subject = cache\n subject.MetricAdd("cache.hit")\n', + ): + with self.subTest( + source_text=source_text + ), tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text(source_text, encoding="utf-8") + with self.assertRaisesRegex(ScriptedGameplayAuditError, "rebound telemetry"): + discover_sites(root) + + def test_guild_factory_alias_is_discovered(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'storage = Guild(name)\nstorage.log_add("fixture")\n', + encoding="utf-8", + ) + _, logs = discover_sites(root) + self.assertEqual("Guild.log_add", logs[0]["facility"]) + + def test_reserved_callable_shadowing_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'def helper(print):\n print("ordinary callback")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "shadows a reserved"): + discover_sites(root) + + def test_guild_receiver_rebinding_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'guild = cache\nguild.log_add("ordinary cache")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "rebinds the guild"): + discover_sites(root) + + def test_print_alias_rebinding_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'emit = print\nemit = callback\nemit("ordinary callback")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "rebinds a print alias"): + discover_sites(root) + + def test_reserved_metric_on_unreviewed_receiver_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'cache.MetricAdd("cache.hit")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "unreviewed receiver"): + discover_sites(root) + + def test_unreviewed_log_add_receiver_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text('self.guild.log_add("fixture")\n', encoding="utf-8") + with self.assertRaisesRegex(ScriptedGameplayAuditError, "indirect audit-log"): + discover_sites(root) + + def test_unbound_getattribute_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'object.__getattribute__(player, "MetricAdd")("hidden")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reflective telemetry"): + discover_sites(root) + + def test_attrgetter_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'operator.attrgetter("MetricAdd")(player)("hidden")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "attribute construction"): + discover_sites(root) + + def test_builtins_dict_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'import builtins\nbuiltins.__dict__["print"]("hidden")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reflective telemetry"): + discover_sites(root) + + def test_wrapped_receiver_reflection_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'getattr((player,)[0], method)("hidden")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reflective telemetry"): + discover_sites(root) + + def test_typed_receiver_vars_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'vars(type(player))["MetricAdd"](player, "hidden")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "reflective telemetry"): + discover_sites(root) + + def test_builtin_namespace_subscription_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + '__builtins__["print"]("hidden")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "namespace access"): + discover_sites(root) + + def test_globals_subscription_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text('globals()["Logger"]("hidden")\n', encoding="utf-8") + with self.assertRaisesRegex(ScriptedGameplayAuditError, "namespace access"): + discover_sites(root) + + def test_imported_logger_alias_shadowing_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text( + 'from Atrinik import Logger as report\ndef run(report):\n report("ordinary")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "shadows a reserved"): + discover_sites(root) + + def test_unreviewed_logger_receiver_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "maps" / "python" / "feature.py" + source.parent.mkdir(parents=True) + source.write_text('cache.Logger("ordinary")\n', encoding="utf-8") + with self.assertRaisesRegex(ScriptedGameplayAuditError, "indirect audit-log"): + discover_sites(root) + + def test_audit_facility_cannot_be_gameplay_journal(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + target = self._fixture( + root, + 'player.MetricAdd("economy.new_action")\nLogger("fixture")\n', + ) + document = json.loads(target.read_text(encoding="utf-8")) + document["audit_like_sites"][0]["classification"] = "gameplay-journal" + target.write_text(json.dumps(document), encoding="utf-8") + with self.assertRaisesRegex(ScriptedGameplayAuditError, "operational or not recorded"): + load_and_validate(root) + + def test_relocated_occurrence_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + self._fixture( + root, + 'player.MetricAdd("economy.new_action")\nLogger("fixture")\n', + ) + source = root / "maps" / "python" / "feature.py" + source.write_text( + 'if committed:\n player.MetricAdd("economy.new_action")\nLogger("fixture")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "inventory differs"): + load_and_validate(root) + + def test_changed_surrounding_operation_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + self._fixture( + root, + 'charge()\nplayer.MetricAdd("economy.new_action")\nLogger("fixture")\n', + ) + source = root / "maps" / "python" / "feature.py" + source.write_text( + 'debit()\nplayer.MetricAdd("economy.new_action")\nLogger("fixture")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "inventory differs"): + load_and_validate(root) + + def test_audit_like_drift_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + self._fixture( + root, + 'player.MetricAdd("economy.new_action")\nLogger("fixture")\n', + ) + source = root / "maps" / "python" / "feature.py" + source.write_text( + 'player.MetricAdd("economy.new_action")\nLogger("fixture")\nLogger("new")\n', + encoding="utf-8", + ) + with self.assertRaisesRegex(ScriptedGameplayAuditError, "inventory differs"): + load_and_validate(root) + + def test_map_local_script_is_discovered(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + self._fixture( + root, + 'player.MetricAdd("economy.new_action")\nLogger("fixture")\n', + "maps/area/scripts/feature.py", + ) + report = load_and_validate(root) + self.assertEqual(1, report["metric_sites"]) + self.assertEqual(1, report["audit_like_sites"]) + + def test_proposed_reason_is_ascii_and_bounded(self): + for invalid in ("player supplied text", "x." + "y" * 254, "fixture.évent"): + with self.subTest(invalid=invalid), tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + target = self._fixture( + root, + 'player.MetricAdd("economy.new_action")\nLogger("fixture")\n', + ) + fixture = json.loads(target.read_text(encoding="utf-8")) + fixture["metric_sites"][0]["proposed_journal_reason"] = invalid + target.write_text(json.dumps(fixture), encoding="utf-8") + with self.assertRaisesRegex( + ScriptedGameplayAuditError, "bounded ASCII|invalid or exceeds" + ): + load_and_validate(root) + + def test_backslash_path_is_rejected(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + target = self._fixture( + root, + 'player.MetricAdd("economy.new_action")\nLogger("fixture")\n', + ) + document = json.loads(target.read_text(encoding="utf-8")) + document["metric_sites"][0]["path"] = "maps/python/..\\outside.py" + target.write_text(json.dumps(document), encoding="utf-8") + with self.assertRaisesRegex(ScriptedGameplayAuditError, "canonical POSIX"): + load_and_validate(root) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/validate.py b/tools/validate.py index 72cc2665d..45e064789 100755 --- a/tools/validate.py +++ b/tools/validate.py @@ -20,6 +20,7 @@ ) from tools.m1_foundations import validate as validate_m1_foundations from tools.release_line_parity import load_and_validate as validate_release_line_parity +from tools.scripted_gameplay_audit import load_and_validate as validate_scripted_gameplay_audit ROOT = Path(__file__).parents[1].resolve() @@ -69,6 +70,7 @@ def main() -> int: "tools.tests.test_release_line_parity", "tools.tests.test_pr_metadata", "tools.tests.test_python_commands", + "tools.tests.test_scripted_gameplay_audit", ], cwd=ROOT, check=True, @@ -76,6 +78,16 @@ def main() -> int: plural_report = audit_archetype_plurals( ROOT, load_manifest(ROOT / MANIFEST_PATH) ) + scripted_audit = validate_scripted_gameplay_audit(ROOT) + print( + "Scripted gameplay audit: {} calls across {} metric identities; " + "{} audit-like facilities classified.".format( + scripted_audit["metric_sites"], + scripted_audit["metric_identities"], + scripted_audit["audit_like_sites"], + ), + flush=True, + ) print( "Archetype plurals: {} canonical definitions complete; {} multipart or " "nested objects excluded.".format( @@ -216,6 +228,12 @@ def main() -> int: or not classic_manifest["license_files"] ): raise ValueError("classic runtime target metadata is invalid") + if ( + classic_output / "contracts" / "scripted-gameplay-audit" / "v1.json" + ).exists() or ( + classic_output / "maps" / "python" / "scripted-gameplay-audit-v1.json" + ).exists(): + raise ValueError("review-only scripted gameplay audit entered runtime") subprocess.run( [