#!/usr/bin/env python3
"""Synthetic file-tool boundary lab. POSIX only; no agent or network calls.

Run with a private --scratch directory. All fixture data is disposable.
The bounded reader accepts one flat filename under a trusted directory fd.
It is a teaching component, not a general filesystem sandbox.
"""
import argparse
import os
from pathlib import Path
import stat
import tempfile
import unittest

LIMIT = 4096


def bounded_read(root_fd, name):
    if not isinstance(name, str) or name in ('', '.', '..'):
        raise ValueError('one filename required')
    if '/' in name or '\\' in name or '\x00' in name:
        raise ValueError('path syntax is forbidden')
    flags = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK
    fd = os.open(name, flags, dir_fd=root_fd)
    try:
        if not stat.S_ISREG(os.fstat(fd).st_mode):
            raise ValueError('regular file required')
        with os.fdopen(fd, 'rb', closefd=False) as stream:
            data = stream.read(LIMIT + 1)
        if len(data) > LIMIT:
            raise ValueError('file exceeds read limit')
        return data
    finally:
        os.close(fd)


def lexical_read(root, name):
    candidate = root / name
    if not candidate.is_relative_to(root):
        raise ValueError('outside root')
    return candidate.read_bytes()


def checked_read(root, name, after_check=lambda: None):
    candidate = (root / name).resolve(strict=True)
    if not candidate.is_relative_to(root.resolve(strict=True)):
        raise ValueError('outside root')
    after_check()  # Deterministic scheduling seam, not a timing benchmark.
    return candidate.read_bytes()


class Fixture:
    def __init__(self, scratch):
        self.tmp = tempfile.TemporaryDirectory(prefix='file-boundary-', dir=scratch)
        self.base = Path(self.tmp.name).resolve()
        self.root = self.base / 'allowed'
        self.root.mkdir()
        self.outside = self.base / 'outside'
        self.outside.mkdir()
        (self.outside / 'note.txt').write_bytes(b'OUTSIDE fixture only\n')
        (self.root / 'note.txt').write_bytes(b'APPROVED fixture only\n')
        (self.root / 'docs').symlink_to(self.outside, target_is_directory=True)
        self.fd = os.open(self.root, os.O_RDONLY | os.O_DIRECTORY)

    def swap(self):
        (self.root / 'note.txt').unlink()
        (self.root / 'note.txt').symlink_to(self.outside / 'note.txt')

    def close(self):
        os.close(self.fd)
        self.tmp.cleanup()


SCRATCH = None


class BoundaryTests(unittest.TestCase):
    def setUp(self):
        self.f = Fixture(SCRATCH)

    def tearDown(self):
        self.f.close()

    def test_lexical_check_follows_directory_link(self):
        self.assertEqual(lexical_read(self.f.root, 'docs/note.txt'), b'OUTSIDE fixture only\n')

    def test_resolve_rejects_existing_escape(self):
        with self.assertRaises(ValueError):
            checked_read(self.f.root, 'docs/note.txt')

    def test_resolve_then_open_has_gap(self):
        self.assertEqual(checked_read(self.f.root, 'note.txt', self.f.swap), b'OUTSIDE fixture only\n')

    def test_bounded_read_accepts_regular_file(self):
        self.assertEqual(bounded_read(self.f.fd, 'note.txt'), b'APPROVED fixture only\n')

    def test_bounded_read_rejects_symlink(self):
        self.f.swap()
        with self.assertRaises(OSError):
            bounded_read(self.f.fd, 'note.txt')

    def test_bounded_read_rejects_path_syntax(self):
        for name in ['', '.', '..', '../outside/note.txt', '/etc/passwd', 'docs/note.txt', 'a\\b', 'a\x00b']:
            with self.subTest(name=name), self.assertRaises(ValueError):
                bounded_read(self.f.fd, name)

    def test_bounded_read_rejects_directory(self):
        (self.f.root / 'dir').mkdir()
        with self.assertRaises(ValueError):
            bounded_read(self.f.fd, 'dir')

    def test_bounded_read_rejects_fifo_without_waiting_for_writer(self):
        os.mkfifo(self.f.root / 'pipe')
        with self.assertRaises(ValueError):
            bounded_read(self.f.fd, 'pipe')

    def test_bounded_read_rejects_oversize(self):
        (self.f.root / 'large').write_bytes(b'x' * (LIMIT + 1))
        with self.assertRaises(ValueError):
            bounded_read(self.f.fd, 'large')

    def test_open_descriptor_survives_name_replacement(self):
        fd = os.open('note.txt', os.O_RDONLY | os.O_NOFOLLOW, dir_fd=self.f.fd)
        try:
            self.f.swap()
            self.assertEqual(os.read(fd, LIMIT), b'APPROVED fixture only\n')
        finally:
            os.close(fd)

    def test_hard_link_is_not_rejected(self):
        os.link(self.f.outside / 'note.txt', self.f.root / 'hardlink')
        self.assertEqual(bounded_read(self.f.fd, 'hardlink'), b'OUTSIDE fixture only\n')


def main():
    global SCRATCH
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--scratch', required=True, type=Path)
    parser.add_argument('--test', action='store_true')
    args = parser.parse_args()
    if os.name != 'posix' or os.open not in os.supports_dir_fd:
        parser.error('POSIX os.open(dir_fd=...) support required')
    for flag in ['O_NOFOLLOW', 'O_NONBLOCK', 'O_DIRECTORY']:
        if not hasattr(os, flag):
            parser.error('missing ' + flag)
    SCRATCH = args.scratch.resolve(strict=True)
    if args.test:
        result = unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(BoundaryTests))
        return 0 if result.wasSuccessful() else 1
    f = Fixture(SCRATCH)
    try:
        print('lexical check:', lexical_read(f.root, 'docs/note.txt').decode().strip())
        print('resolve then swap:', checked_read(f.root, 'note.txt', f.swap).decode().strip())
        try:
            bounded_read(f.fd, 'note.txt')
        except OSError:
            print('bounded reader: symlink rejected')
        else:
            raise RuntimeError('symlink unexpectedly accepted')
    finally:
        f.close()
    return 0


if __name__ == '__main__':
    raise SystemExit(main())