XFE Git
XFE Studio Git
Git 首页 全局搜索
XFE 主站 文档 NuGet

XFESystem

【Rust】XFE系统,基于Rust和汇编

公开
关注 0 Fork 0 Star 0
UTF-8
"""Compile and exercise production CPU entropy as a small no-CRT host DLL.

This executes the current host CPU's RDSEED/RDRAND, not the guest kernel or
TLS. No synthetic entropy implementation or replacement CPU feature is used.
"""
import argparse
import ctypes
import hashlib
import json
from pathlib import Path
import platform
import subprocess
import time


ROOT = Path(__file__).resolve().parents[3]
EXPORTS = ('xfe_runtime_random_capabilities', 'xfe_runtime_random_fill',
           'SystemNative_GetCryptographicallySecureRandomBytes')


def sha256(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def run(output=None, tool_root=None):
    if platform.system() != 'Windows' or ctypes.sizeof(ctypes.c_void_p) != 8:
        raise RuntimeError('This bounded proof requires 64-bit Windows Python')
    source = ROOT / 'Source/XFESystem.NativeRuntime/Native/NativeEntropy.c'
    tool_root = Path(tool_root).resolve() if tool_root else ROOT / '.tools'
    clang, linker, inspector = (tool_root / 'llvm/bin' / name for name in
                                ('clang.exe', 'lld-link.exe', 'llvm-readobj.exe'))
    for dependency in (source, clang, linker, inspector):
        if not dependency.is_file(): raise FileNotFoundError(dependency)
    output = Path(output).resolve() if output else ROOT / 'build/IndependentKernel/EntropyHostProof' / ('Proof-' + str(time.time_ns()))
    if not output.is_relative_to(ROOT / 'build'):
        raise ValueError('Proof outputs must remain below this worktree build directory')
    output.mkdir(parents=True, exist_ok=False)
    source_hash = sha256(source)
    result = {'Kind': 'ProductionNativeEntropyHostProof', 'Passed': False,
              'Source': str(source), 'SourceSha256': source_hash, 'SourceUnchanged': False,
              'Checks': [], 'HostMachine': platform.machine(), 'GuestExecutionVerified': False,
              'GuestTlsVerified': False, 'EntropyQualityCertified': False,
              'HostRandomFallbackUsed': False, 'SyntheticEntropyImplementationUsed': False,
              'Meaning': 'Production CPU instructions and API guards executed on this host; not guest TLS or an entropy-quality certification.'}

    def check(name, condition):
        result['Checks'].append({'Name': name, 'Passed': bool(condition)})
        if not condition: raise AssertionError(name)

    def command(name, arguments):
        completed = subprocess.run([str(value) for value in arguments], capture_output=True,
                                   timeout=30, text=True, encoding='utf-8', errors='replace')
        (output / (name + '.log')).write_text(completed.stdout + completed.stderr, encoding='utf-8')
        result.setdefault('Commands', []).append({'Arguments': [str(value) for value in arguments], 'ExitCode': completed.returncode})
        if completed.returncode: raise RuntimeError(name + ' failed; inspect its bounded log')
        return completed.stdout

    try:
        obj, dll = output / 'NativeEntropy.obj', output / 'NativeEntropy.dll'
        command('Compile', [clang, '--target=x86_64-pc-windows-msvc', '-O2', '-ffreestanding',
                            '-fno-stack-protector', '-fno-unwind-tables', '-fno-asynchronous-unwind-tables',
                            '-c', source, '-o', obj])
        command('Link', [linker, '/dll', '/noentry', '/nodefaultlib', '/machine:x64',
                         '/out:' + str(dll), *('/export:' + name for name in EXPORTS), obj])
        imports = command('Imports', [inspector, '--coff-imports', dll])
        exports = command('Exports', [inspector, '--coff-exports', dll])
        check('NoHostOsOrCrtImports', 'Import {' not in imports)
        check('AllThreeProductionExports', all('Name: ' + name + '\n' in exports for name in EXPORTS))
        result['DllPath'] = str(dll); result['DllSha256'] = sha256(dll)
        result['CompilerSha256'] = sha256(clang); result['LinkerSha256'] = sha256(linker)
        library = ctypes.CDLL(str(dll))
        capabilities = library.xfe_runtime_random_capabilities
        capabilities.argtypes = []; capabilities.restype = ctypes.c_uint32
        fill = library.xfe_runtime_random_fill
        fill.argtypes = [ctypes.POINTER(ctypes.c_uint8), ctypes.c_uint32]; fill.restype = ctypes.c_int
        system_fill = library.SystemNative_GetCryptographicallySecureRandomBytes
        system_fill.argtypes = [ctypes.POINTER(ctypes.c_uint8), ctypes.c_int32]; system_fill.restype = ctypes.c_int32
        flags = capabilities(); result['CpuEntropyCapabilities'] = flags
        check('CapabilitiesContainOnlyRdrandRdseedBits', flags & ~3 == 0)
        check('NativeNullZeroLengthSucceeds', fill(None, 0) == 1)
        check('DotNetNullZeroLengthReturnsZero', system_fill(None, 0) == 0)
        check('NativeNullNonemptyRejected', fill(None, 1) == 0)
        check('DotNetNullNonemptyReturnsMinusOne', system_fill(None, 1) == -1)
        check('DotNetNegativeLengthReturnsMinusOne', system_fill(None, -1) == -1)
        guard = (ctypes.c_uint8 * 32)(*([0xA5] * 32))
        check('NativeOverOneMiBLimitRejectedBeforeAccess', fill(guard, 1_048_577) == 0 and bytes(guard) == b'\xA5' * 32)
        check('DotNetOverOneMiBLimitReturnsMinusOneBeforeAccess', system_fill(guard, 1_048_577) == -1 and bytes(guard) == b'\xA5' * 32)
        check('DotNetNegativeLengthDoesNotWrite', system_fill(guard, -1) == -1 and bytes(guard) == b'\xA5' * 32)
        check('NativeZeroLengthDoesNotWrite', fill(guard, 0) == 1 and bytes(guard) == b'\xA5' * 32)
        check('ActualCpuSourceAdvertised', flags != 0)
        samples = []
        for size in (32, 4096):
            values = []
            for iteration in range(2):
                buffer = (ctypes.c_uint8 * size)()
                check('ActualCpuFill_' + str(size) + '_' + str(iteration), fill(buffer, size) == 1)
                value = bytes(buffer); values.append(value)
                check('NonconstantActualBytes_' + str(size) + '_' + str(iteration), len(set(value)) > 1)
                samples.append({'Bytes': size, 'Sha256': hashlib.sha256(value).hexdigest()})
            check('SeparateActualCallsDiffer_' + str(size), values[0] != values[1])
        buffer = (ctypes.c_uint8 * 32)()
        check('DotNetActualCpuFillReturnsZero', system_fill(buffer, 32) == 0 and len(set(bytes(buffer))) > 1)
        result['Samples'] = samples
        result['SourceUnchanged'] = sha256(source) == source_hash
        check('ProductionSourceUnchanged', result['SourceUnchanged'])
        result['Passed'] = True
    except Exception as error:
        result['Failure'] = type(error).__name__ + ': ' + str(error)
    finally:
        result['SourceUnchanged'] = sha256(source) == source_hash
        result['CheckCount'] = len(result['Checks'])
        (output / 'Result.json').write_text(json.dumps(result, indent=2), encoding='utf-8')
    print(json.dumps({'Passed': result['Passed'], 'CheckCount': result['CheckCount'], 'Output': str(output),
                      'GuestTlsVerified': False, 'Failure': result.get('Failure')}))
    return result


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--output', help='Fresh proof directory below this worktree build directory')
    parser.add_argument('--tool-root', help='Existing tool root containing llvm/bin; never installs dependencies')
    options = parser.parse_args()
    raise SystemExit(0 if run(options.output, options.tool_root)['Passed'] else 1)