-
Notifications
You must be signed in to change notification settings - Fork 0
Add malicious control character coverage #2
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -6,13 +6,26 @@ pythonpath="${git_toplevel}/usr/lib/python3/dist-packages" | |
| export PYTHONPATH="${pythonpath}${PYTHONPATH+":${PYTHONPATH}"}" | ||
|
|
||
| pytest=(python3 -m pytest -o 'python_files=tests/*.py') | ||
| pytest_args=() | ||
| for arg in "$@"; do | ||
| test_path="${pythonpath}/stdisplay/tests/${arg}.py" | ||
| if [[ "${arg}" != -* && "${arg}" != */* && -f "${test_path}" ]]; then | ||
| pytest_args+=("tests/${arg}.py") | ||
| else | ||
| pytest_args+=("${arg}") | ||
| fi | ||
| done | ||
|
|
||
| if [[ ${#pytest_args[@]} -eq 0 ]]; then | ||
| pytest_args=("${@}") | ||
| fi | ||
| black=(black --config="${pyrc}" --color --diff --check) | ||
| pylint=(pylint --rcfile="${pyrc}") | ||
| mypy=(mypy --config-file="${pyrc}") | ||
|
|
||
| cd "${pythonpath}/stdisplay/" | ||
| # Ideally, these variables should be ignored by the tests... | ||
| NO_COLOR="" COLORTERM="" TERM="xterm-direct" "${pytest[@]}" "${@}" | ||
| NO_COLOR="" COLORTERM="" TERM="xterm" "${pytest[@]}" "${pytest_args[@]}" | ||
|
Comment on lines
-15
to
+28
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why not |
||
| "${black[@]}" . | ||
| find . -type f -name "*.py" -print0 | xargs -0 "${pylint[@]}" | ||
| "${mypy[@]}" . | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -7,17 +7,17 @@ | |
|
|
||
| """Safely print argument to stdout with echo's formatting.""" | ||
|
|
||
| from sys import argv, stdout | ||
| import sys | ||
| from stdisplay.stdisplay import stdisplay | ||
|
|
||
|
|
||
| def main() -> None: | ||
| """Safely print argument to stdout with echo's formatting.""" | ||
| if len(argv) > 1: | ||
| untrusted_text = " ".join(argv[1:]) | ||
| stdout.write(stdisplay(untrusted_text)) | ||
| stdout.write("\n") | ||
| stdout.flush() | ||
| if len(sys.argv) > 1: | ||
| untrusted_text = " ".join(sys.argv[1:]) | ||
| sys.stdout.write(stdisplay(untrusted_text)) | ||
| sys.stdout.write("\n") | ||
| sys.stdout.flush() | ||
|
Comment on lines
-16
to
+20
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. There's no reason to change this. |
||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -7,8 +7,8 @@ | |
|
|
||
| """Safely print stdin to stdout or file.""" | ||
|
|
||
| import os | ||
| from sys import argv, stdin, stdout, modules | ||
| import shutil | ||
| from tempfile import NamedTemporaryFile | ||
| from stdisplay.stdisplay import stdisplay | ||
|
|
||
|
|
@@ -17,7 +17,7 @@ def main() -> None: | |
| """Safely print stdin to stdout or file.""" | ||
| # https://github.com/pytest-dev/pytest/issues/4843 | ||
| if "pytest" not in modules and stdin is not None: | ||
| stdin.reconfigure(errors="ignore") # type: ignore | ||
| stdin.reconfigure(errors="replace") # type: ignore | ||
| untrusted_text_list = [] | ||
| if stdin is not None: | ||
| for untrusted_text in stdin: | ||
|
|
@@ -29,9 +29,22 @@ def main() -> None: | |
| with NamedTemporaryFile(mode="w", delete=False) as temp_file: | ||
| temp_file.write(stdisplay("".join(untrusted_text_list))) | ||
| temp_file.flush() | ||
| for file in argv[1:]: | ||
| shutil.copy2(temp_file.name, file) | ||
| temp_file.close() | ||
| temp_path = temp_file.name | ||
| try: | ||
| with open(temp_path, "rb") as source_file: | ||
| content = source_file.read() | ||
| for file in argv[1:]: | ||
| fd = os.open( | ||
| file, | ||
| os.O_CREAT | os.O_WRONLY | os.O_TRUNC | os.O_NOFOLLOW, | ||
| 0o600, | ||
| ) | ||
| with os.fdopen(fd, "wb") as destination_file: | ||
| destination_file.write(content) | ||
| destination_file.flush() | ||
| os.fchmod(destination_file.fileno(), 0o600) | ||
| finally: | ||
| os.unlink(temp_path) | ||
|
Comment on lines
-32
to
+47
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. What's wrong with copy2 here? It's almost certainly faster than doing this manually (thanks to https://docs.python.org/3/library/shutil.html#shutil-platform-dependent-efficient-copy-operations). Furthermore, why are we changing the file permissions here? Isn't whatever the active umask provides good here? I note in similar code below, following symlinks is considered a bad thing, but I disagree; |
||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,7 +5,8 @@ | |
| ## | ||
| ## SPDX-License-Identifier: AGPL-3.0-or-later | ||
|
|
||
| """Safely print stdin to stdout and file.""" | ||
| """Safely print stdin to stdout and file without following symlinks.""" | ||
| import os | ||
|
Comment on lines
-8
to
+9
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
| from sys import argv, stdin, stdout, modules | ||
| from stdisplay.stdisplay import stdisplay | ||
|
|
||
|
|
@@ -14,17 +15,29 @@ def main() -> None: | |
| """Safely print stdin to stdout and file.""" | ||
| # https://github.com/pytest-dev/pytest/issues/4843 | ||
| if "pytest" not in modules and stdin is not None: | ||
| stdin.reconfigure(errors="ignore") # type: ignore | ||
| untrusted_text_list = [] | ||
| if stdin is not None: | ||
| for untrusted_text in stdin: | ||
| untrusted_text_list.append(untrusted_text) | ||
| stdout.write(stdisplay(untrusted_text)) | ||
| stdout.flush() | ||
| if len(argv) > 1: | ||
| for file_arg in argv[1:]: | ||
| with open(file_arg, mode="w", encoding="ascii") as file: | ||
| file.write(stdisplay("".join(untrusted_text_list))) | ||
| stdin.reconfigure(errors="replace") # type: ignore | ||
| output_files = [] | ||
| try: | ||
| if len(argv) > 1: | ||
| for file_arg in argv[1:]: | ||
| fd = os.open( | ||
| file_arg, | ||
| os.O_CREAT | os.O_WRONLY | os.O_TRUNC | os.O_NOFOLLOW, | ||
| 0o600, | ||
| ) | ||
| output_files.append(os.fdopen(fd, mode="w", encoding="ascii")) | ||
| if stdin is not None: | ||
| for untrusted_text in stdin: | ||
| rendered_text = stdisplay(untrusted_text) | ||
| stdout.write(rendered_text) | ||
| for output_file in output_files: | ||
| output_file.write(rendered_text) | ||
| stdout.flush() | ||
| for output_file in output_files: | ||
| output_file.flush() | ||
| finally: | ||
| for output_file in output_files: | ||
| output_file.close() | ||
|
Comment on lines
-17
to
+40
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I like the memory optimization and multiple streaming here. I don't agree with the "don't follow symlinks" approach, but otherwise the basic idea here seems useful. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This will effectively implement a fix for Kicksecure#18. Will integrate. |
||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -34,6 +34,9 @@ class TestSTBase(unittest.TestCase): | |
| def __init__(self, *args: Any, **kwargs: Any) -> None: | ||
| self.text_dirty = "\x1b[0mTest\x1b[2Kor\x1b]1;is\x1b\n[m" | ||
| self.text_dirty_sanitized = "\x1b[0mTest_[2Kor_]1;is_\n[m" | ||
| self.text_malicious = "pre\u202e.js\u200b\x00post\n" | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Right-to-left override (U+202e) is interesting, but it's been known about for a bit longer than the Trojan Source disclosure, and isn't as concerning as the use of isolate shuffling. I would argue we should use isolate shuffling here instead. |
||
| self.text_malicious_sanitized = "pre_.js__post\n" | ||
| self.text_malicious_file_sanitized = "pre___.js____post\n" | ||
|
Comment on lines
+37
to
+39
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why... are there two different sanitized end results for sanitizing a string vs sanitizing a file? These should have the exact same result. |
||
| super().__init__(*args, **kwargs) | ||
|
|
||
| def setUp(self) -> None: | ||
|
|
@@ -64,6 +67,20 @@ def setUp(self) -> None: | |
| "fill2": self.tmpfiles_list[5], | ||
| } | ||
|
|
||
| invalid_path = os.path.join(self.tmpdir, "invalid") | ||
| with open(invalid_path, "wb") as file: | ||
| file.write(b"a\xffb\n") | ||
| file.flush() | ||
| file.close() | ||
| self.tmpfiles["invalid"] = invalid_path | ||
|
|
||
| malicious_path = os.path.join(self.tmpdir, "malicious") | ||
| with open(malicious_path, "w", encoding="utf-8") as file: | ||
| file.write(self.text_malicious) | ||
| file.flush() | ||
| file.close() | ||
| self.tmpfiles["malicious"] = malicious_path | ||
|
|
||
| def tearDown(self) -> None: | ||
| shutil.rmtree(self.tmpdir) | ||
|
|
||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -6,8 +6,12 @@ | |
| ## | ||
| ## SPDX-License-Identifier: AGPL-3.0-or-later | ||
|
|
||
| import importlib | ||
| import io | ||
| import sys | ||
| import unittest | ||
| from pathlib import Path | ||
| from unittest.mock import patch | ||
| import stdisplay.tests | ||
|
|
||
|
|
||
|
|
@@ -61,6 +65,68 @@ def test_stsponge(self) -> None: | |
| Path(self.tmpfiles["fill2"]).read_text(encoding="utf-8"), | ||
| ) | ||
|
|
||
| def test_stsponge_replaces_invalid_bytes(self) -> None: | ||
| """Invalid bytes are sanitized for stdout and files.""" | ||
|
|
||
| output_path = Path(self.tmpfiles["fill"]) | ||
| invalid_bytes = b"a\xffb\n" | ||
| stdout_capture = io.StringIO() | ||
| pytest_module = sys.modules.pop("pytest", None) | ||
| try: | ||
| with patch.object(sys, "argv", ["stsponge.py", str(output_path)]), patch( | ||
| "sys.stdin", | ||
| io.TextIOWrapper(io.BytesIO(invalid_bytes), encoding="utf-8"), | ||
| ), patch("sys.stdout", stdout_capture): | ||
| self._del_module() | ||
| importlib.import_module("stdisplay.stsponge").main() | ||
| finally: | ||
| if pytest_module is not None: | ||
| sys.modules["pytest"] = pytest_module | ||
|
Comment on lines
+74
to
+84
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We have |
||
|
|
||
| self.assertEqual("", stdout_capture.getvalue()) | ||
| self.assertEqual("a_b\n", output_path.read_text(encoding="ascii")) | ||
|
|
||
| stdout_capture = io.StringIO() | ||
| pytest_module = sys.modules.pop("pytest", None) | ||
| try: | ||
| with patch.object(sys, "argv", ["stsponge.py"]), patch( | ||
| "sys.stdin", | ||
| io.TextIOWrapper(io.BytesIO(invalid_bytes), encoding="utf-8"), | ||
| ), patch("sys.stdout", stdout_capture): | ||
| self._del_module() | ||
| importlib.import_module("stdisplay.stsponge").main() | ||
| finally: | ||
| if pytest_module is not None: | ||
| sys.modules["pytest"] = pytest_module | ||
|
|
||
| self.assertEqual("a_b\n", stdout_capture.getvalue()) | ||
|
|
||
| def test_stsponge_sanitizes_control_and_bidi(self) -> None: | ||
| """Control and bidi characters are sanitized in outputs.""" | ||
|
|
||
| output_path = Path(self.tmpfiles["fill"]) | ||
| stdout_capture = io.StringIO() | ||
| with patch.object(sys, "argv", ["stsponge.py", str(output_path)]), patch( | ||
| "sys.stdin", io.StringIO(self.text_malicious) | ||
| ), patch("sys.stdout", stdout_capture): | ||
| self._del_module() | ||
| importlib.import_module("stdisplay.stsponge").main() | ||
|
|
||
| self.assertEqual("", stdout_capture.getvalue()) | ||
| self.assertEqual( | ||
| self.text_malicious_sanitized, | ||
| output_path.read_text(encoding="utf-8"), | ||
| ) | ||
|
|
||
| stdout_capture = io.StringIO() | ||
| with patch.object(sys, "argv", ["stsponge.py"]), patch( | ||
| "sys.stdin", io.StringIO(self.text_malicious) | ||
| ), patch("sys.stdout", stdout_capture): | ||
| self._del_module() | ||
| importlib.import_module("stdisplay.stsponge").main() | ||
|
|
||
| self.assertEqual(self.text_malicious_sanitized, stdout_capture.getvalue()) | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| unittest.main() | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't see why this is needed. It changes the API of
run-testsfor no apparent reason (except maybe greater convenience? I personally don't see much convenience benefit here, especially if one wants to run tests for more than juststdisplay). IMO this is undesirable.