Skip to content
Merged
19 changes: 18 additions & 1 deletion gitgalaxy/core/aperture.py
Original file line number Diff line number Diff line change
Expand Up @@ -402,8 +402,12 @@ def _check_artifact_integrity(
# 15-318 KB of text were dropped whole). Every binary in the pinned crucible
# carries 178+ NULs per 1000 characters (tar, .mat); text damaged in transfer
# carries a handful. More than 1 per 1000 is binary.
# #4351: ... unless the member is short and the NULs are a handful: a 168-byte PL/I member
# with 2 NULs at the end of a comment (estate-crucible NULREST.pli, H-0037) is 12 per 1000.
# At most _STRAY_NULS NULs, with no more than as many other control characters, is
# transfer damage in text, whatever the size; a binary carries control bytes beyond its NULs.
nuls = content.count("\x00") if "\x00" in content else 0
if nuls and nuls * 1000 > len(content):
if nuls and nuls * 1000 > len(content) and not self._stray_nuls(content, nuls):
report.update(
{
"valid": False,
Expand Down Expand Up @@ -582,6 +586,19 @@ def _check_artifact_integrity(

return report

# #4351: the most NULs (and, separately, other C0 control characters) a member may carry and
# still be text damaged in transfer rather than a binary.
_STRAY_NULS = 4
_CONTROL_CHARS = re.compile(r"[\x01-\x08\x0e-\x1f\x7f]")

@classmethod
def _stray_nuls(cls, content: str, nuls: int) -> bool:
"""#4351: a handful of NULs in a member that is otherwise text (see Gate 3.1)."""
if nuls > cls._STRAY_NULS:
return False
# stops at the first control character past the limit
return all(n <= cls._STRAY_NULS for n, _ in enumerate(cls._CONTROL_CHARS.finditer(content), 1))

def _check_ignore_rules(self, rel_path: str, has_intent: bool = False) -> bool:
"""
Determines if the path is in a blocked, ignored, or dynamically
Expand Down
32 changes: 22 additions & 10 deletions gitgalaxy/core/data_moves.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,13 +59,24 @@
from typing import Any, Optional

from gitgalaxy.core.db2_declare_table import _blank_sequence_fields
from gitgalaxy.core.special_names import special_names
from gitgalaxy.standards.language_standards.identifiers import NATIONAL, WIDE_DIGITS, WIDE_HYPHENS

_LITERAL = r"[XNGZ]?'[^'\n]{0,320}'?|[XNGZ]?\"[^\"\n]{0,320}\"?"
_NUMBER = r"[+-]?[0-9]*\.[0-9]+|[+-]?[0-9]+"
_WORD = r"[A-Z0-9][A-Z0-9-]{0,62}"
_NUMBER_TOKEN = rf"(?:{_NUMBER})(?![A-Z0-9-])" # a number, not the head of a name (1ST-X)
# #4353: a COBOL word takes the record reader's national, CJK and full-width characters (#3955 /
# #3991): `MOVE '000001' TO 社員コード` drew no row and `X項目` was cut to `X`.
_NAME_START = "A-Z" + NATIONAL + WIDE_DIGITS + "0-9"
_NAME_CHAR = _NAME_START + WIDE_HYPHENS + "-"
_WORD = rf"[{_NAME_START}][{_NAME_CHAR}]{{0,62}}"
_HAS_LETTER = re.compile(f"[A-Z{NATIONAL}]", re.I) # a name has a letter (national ones too, #4353)
_NUMBER_TOKEN = rf"(?:{_NUMBER})(?![{_NAME_CHAR}])" # a number, not the head of a name (1ST-X)
_OPERATOR = r"\*\*|[()=:+*/,.<>-]"
_TOKEN = re.compile("|".join((_LITERAL, _NUMBER_TOKEN, _WORD, _OPERATOR)), re.I)
# #4355: under DECIMAL-POINT IS COMMA, `0,5` is one numeric literal (a separator comma is followed by
# a space). The tokenizer split it into `0` `,` `5`, and `MOVE 0,5 TO T-WERT` drew no row.
_NUMBER_COMMA = r"[+-]?[0-9]*,[0-9]+|" + _NUMBER
_TOKEN_COMMA = re.compile("|".join((_LITERAL, rf"(?:{_NUMBER_COMMA})(?![{_NAME_CHAR}])", _WORD, _OPERATOR)), re.I)
_STATEMENT_TOKENS = 600
_VERBS = frozenset(
{
Expand Down Expand Up @@ -124,8 +135,8 @@ def _is_literal(raw: str) -> bool:


class _Stream:
def __init__(self, toks: list[tuple[str, int]], start: int, end: int):
self.toks, self.i, self.end = toks, start, end
def __init__(self, toks: list[tuple[str, int]], start: int, end: int, number: str = _NUMBER):
self.toks, self.i, self.end, self.number = toks, start, end, number

def peek(self, k: int = 0) -> str:
j = self.i + k
Expand All @@ -146,7 +157,7 @@ def operand(self) -> Optional[tuple[str, str, Any]]:
if _is_literal(raw):
self.i += 1
return raw, "literal", False
if re.fullmatch(_NUMBER, raw):
if re.fullmatch(self.number, raw):
self.i += 1
return raw, "literal", False
# `ALL 'x'`, `ALL X'00'` (#4205: a hexadecimal literal, IBM DBB EPSCSMRD) or `ALL SPACES`.
Expand Down Expand Up @@ -180,7 +191,7 @@ def operand(self) -> Optional[tuple[str, str, Any]]:
self._skip_parens()
arg = "".join(tok[0] for tok in self.toks[start + 1 : self.i - 1]).upper()
return f"{t}({arg})", "cics_constant", False
if not re.fullmatch(_WORD, raw, re.I) or not re.search(r"[A-Z]", t):
if not re.fullmatch(_WORD, raw, re.I) or not _HAS_LETTER.search(t):
return None
name = t
self.i += 1
Expand Down Expand Up @@ -235,7 +246,7 @@ def expression_items(self) -> list[tuple[str, str, bool]]:
if t == "FUNCTION":
self.i += 2
continue
if t in _NOISE or not re.search(r"[A-Z]", t) or t in _FIGURATIVE:
if t in _NOISE or not _HAS_LETTER.search(t) or t in _FIGURATIVE:
self.i += 1
continue
op = self.operand()
Expand Down Expand Up @@ -377,7 +388,8 @@ def data_moves(code_stream: str) -> list[dict[str, Any]]:
return []
text = _procedure_text(code_stream)
newlines = [i for i, ch in enumerate(text) if ch == "\n"]
toks = [(m.group(0), m.start()) for m in _TOKEN.finditer(text)]
comma = any(sn["clause"] == "DECIMAL-POINT" for sn in special_names(code_stream)) # #4355
toks = [(m.group(0), m.start()) for m in (_TOKEN_COMMA if comma else _TOKEN).finditer(text)]
rows: list[dict[str, Any]] = []
i = 0
while i < len(toks):
Expand All @@ -391,7 +403,7 @@ def data_moves(code_stream: str) -> list[dict[str, Any]]:
i += 1
continue
line = bisect.bisect_left(newlines, toks[i][1]) + 1
s = _Stream(toks, i + 1, min(len(toks), i + 1 + _STATEMENT_TOKENS))
s = _Stream(toks, i + 1, min(len(toks), i + 1 + _STATEMENT_TOKENS), _NUMBER_COMMA if comma else _NUMBER)
for src, target, corr in _rows_of(word, s):
rows.append(
{
Expand Down Expand Up @@ -475,7 +487,7 @@ def rounding_facts(code_stream: str) -> list[dict[str, Any]]:
elif t == "SIZE" and j + 1 < end and toks[j + 1][0].upper() == "ERROR":
size_error = True
break # what follows is the imperative statement run on SIZE ERROR
elif re.fullmatch(_WORD, t, re.I) and re.search(r"[A-Z]", t) and t not in _NOISE and t not in _STOPS:
elif re.fullmatch(_WORD, t, re.I) and _HAS_LETTER.search(t) and t not in _NOISE and t not in _STOPS:
last_name = toks[j][0]
j += 1
if rounded or size_error:
Expand Down
4 changes: 3 additions & 1 deletion gitgalaxy/core/detector.py
Original file line number Diff line number Diff line change
Expand Up @@ -10178,7 +10178,9 @@ def _calculate_block_metrics(
dict.fromkeys(
c
for c in raw_calls
if c not in _CALLS_OUT_GLOBAL_IGNORE
# #4305: a pattern alternative that only consumes text (COBOL `EXIT PERFORM`) captures ""
if c
and c not in _CALLS_OUT_GLOBAL_IGNORE
and (c.casefold() if _fold_ignore else c) not in lang_ignore
and c != self_name
)
Expand Down
17 changes: 15 additions & 2 deletions gitgalaxy/standards/language_standards/languages/cobol.py
Original file line number Diff line number Diff line change
Expand Up @@ -900,12 +900,21 @@
# #3359: `(?<![\w-])`, not `\b` -- the scope terminators `END-PERFORM` /
# `END-CALL` end in the verb, so `\b` let the NEXT statement's first word
# (`END-PERFORM` newline `MOVE ...`) be captured as a callee.
# #4305: three inline PERFORM forms name no paragraph either. `EXIT PERFORM [CYCLE]` leaves an
# inline loop: it is consumed by its own alternative, which captures nothing (an empty callee,
# dropped by the detector). `PERFORM WS-N TIMES` / `PERFORM 3 TIMES` count an inline loop: a
# name followed by TIMES is the count, while `PERFORM PARA-X 3 TIMES` keeps PARA-X. The name
# must end at a word end, so backtracking can never shorten `WS-N` to `WS-` to dodge TIMES.
"calls_out": re.compile(
r"(?i)(?<![\w-])(?:PERFORM|CALL)\s+['\"]?([A-Z"
r"(?i)(?<![\w-])(?:EXIT\s+PERFORM(?:\s+CYCLE)?(?![\w-])|(?:PERFORM|CALL)\s+['\"]?([A-Z"
+ NATIONAL
+ WIDE_DIGITS
+ WIDE_HYPHENS
+ r"a-z0-9_-]+)['\"]?"
+ r"a-z0-9_-]+)(?![A-Z"
+ NATIONAL
+ WIDE_DIGITS
+ WIDE_HYPHENS
+ r"a-z0-9_-])(?!\s+TIMES(?![\w-]))['\"]?)"
),
# #3393: in `CALL 'SUBPROG'` the literal IS the callee (as JCL's PGM=
# is), but the literal shield blanked it before calls_out ran, so only
Expand All @@ -926,6 +935,10 @@
"varying",
"until",
"with",
# #4305: `PERFORM TEST BEFORE|AFTER ...` (WITH omitted) and GnuCOBOL / Micro Focus
# `PERFORM FOREVER` are inline loops too.
"test",
"forever",
}
),
# #3362: GO TO <paragraph|section> -- an unconditional transfer of control.
Expand Down
15 changes: 3 additions & 12 deletions tests/cobol_mainframe/ground_truth_ledger.json
Original file line number Diff line number Diff line change
Expand Up @@ -11,11 +11,6 @@
"kind": "deliberate",
"summary": "A GO TO back to the paragraph it is in (a read loop) is not recorded: the engine drops a unit's transfer to itself like recursion (detector, #3362 contract). The key keeps it, since it is a real edge."
},
"engine_inline_perform_phantom_callee": {
"issue": 4305,
"kind": "defect",
"summary": "An inline PERFORM form (PERFORM TEST AFTER / WITH TEST ...) is read as an out-of-line PERFORM of a paragraph named TEST."
},
"engine_main_line_in_no_unit": {
"issue": 4302,
"kind": "defect",
Expand Down Expand Up @@ -1112,7 +1107,6 @@
"engine | dead | src/base/cobol_src/UPDCUST.cbl | PTD999 | fn": "engine_dead_is_unreferenced_by_name",
"engine | dead | src/base/cobol_src/XFRFUN.cbl | A999 | fn": "engine_dead_is_unreferenced_by_name",
"engine | dead | src/base/cobol_src/XFRFUN.cbl | GMOOH999 | fn": "engine_dead_is_unreferenced_by_name",
"engine | intra-program edges | src/base/cobol_src/BANKDATA.cbl | A010 -> PERFORM TEST | fp": "engine_inline_perform_phantom_callee",
"engine | unit extents | src/base/cobol_src/BANKDATA.cbl | CDW010 L1407-1433 | fn": "engine_unit_end_counts_noncode_lines",
"engine | unit extents | src/base/cobol_src/BANKDATA.cbl | CDW010 L1407-1436 | fp": "engine_unit_end_counts_noncode_lines"
},
Expand Down Expand Up @@ -1454,7 +1448,7 @@
},
"intra-program edges": {
"engine": {
"got": 374,
"got": 373,
"tp": 373,
"truth": 373
},
Expand Down Expand Up @@ -3418,10 +3412,7 @@
"engine | dead | COBOL/SAM1.cbl | 000-MAIN | fp": "engine_dead_is_unreferenced_by_name",
"engine | dead | COBOL/SAM1LIB.cbl | 000-MAIN | fp": "engine_dead_is_unreferenced_by_name",
"engine | dead | COBOL/SAM2.cbl | 000-MAIN | fp": "engine_dead_is_unreferenced_by_name",
"engine | dead | multiroot/sam/SAM2.cbl | 000-MAIN | fp": "engine_dead_is_unreferenced_by_name",
"engine | intra-program edges | COBOL/SAM1.cbl | 210-PROCESS-ADD-TRAN -> PERFORM TEST | fp": "engine_inline_perform_phantom_callee",
"engine | intra-program edges | COBOL/SAM1LIB.cbl | 210-PROCESS-ADD-TRAN -> PERFORM TEST | fp": "engine_inline_perform_phantom_callee",
"engine | intra-program edges | multiroot/sam/SAM1.cbl | 210-PROCESS-ADD-TRAN -> PERFORM TEST | fp": "engine_inline_perform_phantom_callee"
"engine | dead | multiroot/sam/SAM2.cbl | 000-MAIN | fp": "engine_dead_is_unreferenced_by_name"
},
"scoreboard": {
"BMS screen fields": {
Expand Down Expand Up @@ -3737,7 +3728,7 @@
},
"intra-program edges": {
"engine": {
"got": 96,
"got": 93,
"tp": 93,
"truth": 93
},
Expand Down
31 changes: 31 additions & 0 deletions tests/cobol_mainframe/test_data_moves_decimal_comma_4355.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
"""#4355: under DECIMAL-POINT IS COMMA a numeric literal is written `0,5`. data_moves tokenized it
as `0` `,` `5`, so `MOVE 0,5 TO T-WERT (2)` drew no row (estate-crucible ZINSBER.cbl:26, H-0043).
The literal is one token in such a program; a separator comma (followed by a space) still separates,
and a program without the clause is read as before."""

from gitgalaxy.core.data_moves import data_moves

HEAD = (
" IDENTIFICATION DIVISION.\n PROGRAM-ID. ZINSBER.\n ENVIRONMENT DIVISION.\n"
" CONFIGURATION SECTION.\n SPECIAL-NAMES.\n DECIMAL-POINT IS COMMA.\n"
" DATA DIVISION.\n WORKING-STORAGE SECTION.\n 01 T-WERT PIC 9V9 OCCURS 3.\n"
" 01 A PIC 9V99.\n 01 B PIC 9V99.\n PROCEDURE DIVISION.\n"
)
BODY = " MOVE 0,5 TO T-WERT (2)\n MOVE 1,25 TO A, B\n MOVE A TO B.\n"


def _rows(src: str) -> list:
return [(m["verb"], m["source"], m["source_kind"], m["target"]) for m in data_moves(src)]


def test_a_decimal_comma_literal_is_one_source():
rows = _rows(HEAD + BODY)
assert rows[0][:3] == ("MOVE", "0,5", "literal") and rows[0][3].startswith("T-WERT")
# `, ` after the literal is a separator: both targets still get the move
assert [r[3] for r in rows if r[1] == "1,25"] == ["A", "B"]
assert ("MOVE", "A", "item", "B") in rows


def test_without_the_clause_a_comma_still_separates():
plain = HEAD.replace(" DECIMAL-POINT IS COMMA.\n", "")
assert not any(r[1] in ("0,5", "1,25") for r in _rows(plain + BODY))
40 changes: 40 additions & 0 deletions tests/cobol_mainframe/test_data_moves_national_4353.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
"""#4353: data_moves' word class lacked the national, CJK and full-width characters the record
reader takes (#3955 / #3991), and a name had to contain an ASCII letter. `MOVE '000001' TO
社員コード` and `MOVE 'Z' TO F02` drew no row, and `MOVE SPACE TO X項目` was recorded into `X`
(estate-crucible H-0040, H-0041, H-0046, part of H-0042)."""

from gitgalaxy.core.data_moves import data_moves

SRC = """\
PROCEDURE DIVISION.
MOVE '000001' TO 社員コード
MOVE 'Z' TO F02
MOVE SPACE TO X項目
MOVE SPACES TO TEST-DATA1
MOVE 見出し TO 社員番号
ADD 1000 TO 賞与額
MOVE 1 TO WS-COUNT
MOVE 'A' TO WS-X.
"""


def _moves() -> list:
return [(m["verb"], m["source"], m["target"]) for m in data_moves(SRC)]


def test_japanese_full_width_and_mixed_names_are_whole_operands():
assert _moves() == [
("MOVE", "'000001'", "社員コード"),
("MOVE", "'Z'", "F02"),
("MOVE", "SPACE", "X項目"),
("MOVE", "SPACES", "TEST-DATA1"),
("MOVE", "見出し", "社員番号"),
("ADD", "1000", "賞与額"),
("MOVE", "1", "WS-COUNT"),
("MOVE", "'A'", "WS-X"),
]


def test_a_full_width_digit_run_is_not_a_name():
# a name still needs a letter: a bare full-width number is not a data name
assert [m["target"] for m in data_moves(" PROCEDURE DIVISION.\n MOVE 1 TO 12.\n")] == []
31 changes: 30 additions & 1 deletion tests/core_engine/test_aperture.py
Original file line number Diff line number Diff line change
Expand Up @@ -325,14 +325,43 @@ def test_aperture_admits_text_with_a_stray_nul(filter_engine, tmp_path):
assert "Binary Format Detected" not in (result.get("reason") or "")


def test_a_short_member_with_a_few_stray_nuls_stays_source_4351(filter_engine, tmp_path):
"""#4351 (estate-crucible H-0037): a 168-byte PL/I member with two NULs at the end of its
first comment is 12 NULs per 1000 characters -- over the density bar -- yet it is text."""
src = tmp_path / "nulrest.py" # the fixture registry's extension; the gate is extension-blind
content = (
" /* NULREST - RESTSALDO. OVERFOERT MED FTP (BINAER).\x00\x00 */\n"
" NULREST: PROC OPTIONS(MAIN);\n DCL SALDO FIXED DEC(9,2);\n SALDO = 0;\n END NULREST;\n"
)
src.write_text(content, encoding="utf-8")
result = filter_engine.is_in_scope(src, content=content)
assert "Binary Format Detected" not in (result.get("reason") or "")
# five NULs is more than a handful: dense enough, it is binary again
content5 = content.replace("\x00\x00", "\x00\x00\x00\x00\x00")
assert "Binary Format Detected" in (filter_engine.is_in_scope(src, content=content5).get("reason") or "")


def test_a_short_binary_with_few_nuls_is_still_blocked_4351(filter_engine, tmp_path):
"""#4351: a binary blob carries control bytes beyond its NULs, so a handful of NULs alone
does not make it text."""
blob = tmp_path / "small_dat.py"
content = (
"".join(chr(b) for b in (0x7F, 0x45, 0x4C, 0x46, 0x02, 0x01, 0x01, 0x00, 0x03, 0x05, 0x10, 0x1B, 0x06)) * 3
)
content = content.replace("\x00", "", 1) + "\x00"
blob.write_text(content, encoding="utf-8")
assert "Binary Format Detected" in (filter_engine.is_in_scope(blob, content=content).get("reason") or "")


# ==============================================================================
def test_aperture_binary_and_monolith_shields(filter_engine, tmp_path):
"""
Ensures opaque binaries (null bytes) and >30,000 LOC monoliths are blocked.
"""
# 1. Opaque Binary Detected
bin_file = tmp_path / "script.py"
content = "print('hello')\x00\x00"
# #4351: more than a handful of NULs -- two alone in short text are transfer damage, not a binary
content = "print('hello')\x00\x00\x00\x00\x00"
bin_file.write_text(content, encoding="utf-8")

result = filter_engine.is_in_scope(bin_file, content=content)
Expand Down
Loading
Loading