byp4a55 professor6T9

Linux server326.web-hosting.com 4.18.0-513.18.1.lve.el8.x86_64 #1 SMP Thu Feb 22 12:55:50 UTC 2024 x86_64
LiteSpeed
67.223.118.122

File Name : windows_utils.py

"""Various Windows specific bits and pieces."""

import sys

if sys.platform != 'win32':  # pragma: no cover
    raise ImportError('win32 only')

import _winapi
import itertools
import msvcrt
import os
import subprocess
import tempfile
import warnings

__all__ = 'pipe', 'Popen', 'PIPE', 'PipeHandle'

# Constants/globals

BUFSIZE = 8192
PIPE = subprocess.PIPE
STDOUT = subprocess.STDOUT
_mmap_counter = itertools.count()

# Replacement for os.pipe() using handles instead of fds

def pipe(*, duplex=False, overlapped=(True, True), bufsize=BUFSIZE):
    """Like os.pipe() but with overlapped support and using handles not fds."""
    address = tempfile.mktemp(
        prefix=r'\\.\pipe\python-pipe-{:d}-{:d}-'.format(
            os.getpid(), next(_mmap_counter)))

if duplex:
        openmode = _winapi.PIPE_ACCESS_DUPLEX
        access = _winapi.GENERIC_READ | _winapi.GENERIC_WRITE
        obsize, ibsize = bufsize, bufsize
    else:
        openmode = _winapi.PIPE_ACCESS_INBOUND
        access = _winapi.GENERIC_WRITE
        obsize, ibsize = 0, bufsize

openmode |= _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE

if overlapped[0]:
        openmode |= _winapi.FILE_FLAG_OVERLAPPED

if overlapped[1]:
        flags_and_attribs = _winapi.FILE_FLAG_OVERLAPPED
    else:
        flags_and_attribs = 0

h1 = h2 = None
    try:
        h1 = _winapi.CreateNamedPipe(
            address, openmode, _winapi.PIPE_WAIT,
            1, obsize, ibsize, _winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL)

h2 = _winapi.CreateFile(
            address, access, 0, _winapi.NULL, _winapi.OPEN_EXISTING,
            flags_and_attribs, _winapi.NULL)

ov = _winapi.ConnectNamedPipe(h1, overlapped=True)
        ov.GetOverlappedResult(True)
        return h1, h2
    except:
        if h1 is not None:
            _winapi.CloseHandle(h1)
        if h2 is not None:
            _winapi.CloseHandle(h2)
        raise

# Wrapper for a pipe handle

class PipeHandle:
    """Wrapper for an overlapped pipe handle which is vaguely file-object like.

The IOCP event loop can use these instead of socket objects.
    """
    def __init__(self, handle):
        self._handle = handle

def __repr__(self):
        if self._handle is not None:
            handle = f'handle={self._handle!r}'
        else:
            handle = 'closed'
        return f'<{self.__class__.__name__} {handle}>'

@property
    def handle(self):
        return self._handle

def fileno(self):
        if self._handle is None:
            raise ValueError("I/O operation on closed pipe")
        return self._handle

def close(self, *, CloseHandle=_winapi.CloseHandle):
        if self._handle is not None:
            CloseHandle(self._handle)
            self._handle = None

def __del__(self, _warn=warnings.warn):
        if self._handle is not None:
            _warn(f"unclosed {self!r}", ResourceWarning, source=self)
            self.close()

def __enter__(self):
        return self

def __exit__(self, t, v, tb):
        self.close()

# Replacement for subprocess.Popen using overlapped pipe handles

class Popen(subprocess.Popen):
    """Replacement for subprocess.Popen using overlapped pipe handles.

The stdin, stdout, stderr are None or instances of PipeHandle.
    """
    def __init__(self, args, stdin=None, stdout=None, stderr=None, **kwds):
        assert not kwds.get('universal_newlines')
        assert kwds.get('bufsize', 0) == 0
        stdin_rfd = stdout_wfd = stderr_wfd = None
        stdin_wh = stdout_rh = stderr_rh = None
        if stdin == PIPE:
            stdin_rh, stdin_wh = pipe(overlapped=(False, True), duplex=True)
            stdin_rfd = msvcrt.open_osfhandle(stdin_rh, os.O_RDONLY)
        else:
            stdin_rfd = stdin
        if stdout == PIPE:
            stdout_rh, stdout_wh = pipe(overlapped=(True, False))
            stdout_wfd = msvcrt.open_osfhandle(stdout_wh, 0)
        else:
            stdout_wfd = stdout
        if stderr == PIPE:
            stderr_rh, stderr_wh = pipe(overlapped=(True, False))
            stderr_wfd = msvcrt.open_osfhandle(stderr_wh, 0)
        elif stderr == STDOUT:
            stderr_wfd = stdout_wfd
        else:
            stderr_wfd = stderr
        try:
            super().__init__(args, stdin=stdin_rfd, stdout=stdout_wfd,
                             stderr=stderr_wfd, **kwds)
        except:
            for h in (stdin_wh, stdout_rh, stderr_rh):
                if h is not None:
                    _winapi.CloseHandle(h)
            raise
        else:
            if stdin_wh is not None:
                self.stdin = PipeHandle(stdin_wh)
            if stdout_rh is not None:
                self.stdout = PipeHandle(stdout_rh)
            if stderr_rh is not None:
                self.stderr = PipeHandle(stderr_rh)
        finally:
            if stdin == PIPE:
                os.close(stdin_rfd)
            if stdout == PIPE:
                os.close(stdout_wfd)
            if stderr == PIPE:
                os.close(stderr_wfd)

Name	Type	Size	Owner/Group	Permission	Last Modified	Actions
.	dir	-	root/linksafe	0755	2026-02-10 08:06:17
..	dir	-	root/linksafe	0755	2026-02-10 08:06:17
__pycache__	dir	-	root/linksafe	0755	2026-02-10 08:06:17
__init__.py	text/plain	1.19 KB	root/linksafe	0644	2026-01-08 06:12:54
__main__.py	text/x-script.python	3.41 KB	root/linksafe	0644	2026-01-08 06:12:55
base_events.py	text/plain	76.73 KB	root/linksafe	0644	2026-01-08 06:12:55
base_futures.py	text/x-script.python	1.93 KB	root/linksafe	0644	2026-01-08 06:12:55
base_subprocess.py	text/x-script.python	8.66 KB	root/linksafe	0644	2026-01-08 06:12:54
base_tasks.py	text/x-script.python	2.61 KB	root/linksafe	0644	2026-01-08 06:12:54
constants.py	text/x-script.python	1.38 KB	root/linksafe	0644	2026-01-08 06:12:54
coroutines.py	text/x-script.python	3.26 KB	root/linksafe	0644	2026-01-08 06:12:54
events.py	text/plain	28.65 KB	root/linksafe	0644	2026-01-08 06:12:54
exceptions.py	text/plain	1.71 KB	root/linksafe	0644	2026-01-08 06:12:54
format_helpers.py	text/x-script.python	2.35 KB	root/linksafe	0644	2026-01-08 06:12:55
futures.py	text/plain	14 KB	root/linksafe	0644	2026-01-08 06:12:55
locks.py	text/plain	18.55 KB	root/linksafe	0644	2026-01-08 06:12:55
log.py	text/plain	124 B	root/linksafe	0644	2026-01-08 06:12:55
mixins.py	text/plain	481 B	root/linksafe	0644	2026-01-08 06:12:54
proactor_events.py	text/plain	32.71 KB	root/linksafe	0644	2026-01-08 06:12:54
protocols.py	text/plain	6.79 KB	root/linksafe	0644	2026-01-08 06:12:55
queues.py	text/x-script.python	7.79 KB	root/linksafe	0644	2026-01-08 06:12:54
runners.py	text/x-script.python	7.06 KB	root/linksafe	0644	2026-01-08 06:12:54
selector_events.py	text/plain	47.2 KB	root/linksafe	0644	2026-01-08 06:12:54
sslproto.py	text/x-script.python	31.15 KB	root/linksafe	0644	2026-01-08 06:12:55
staggered.py	text/plain	6.91 KB	root/linksafe	0644	2026-01-08 06:12:54
streams.py	text/x-script.python	26.97 KB	root/linksafe	0644	2026-01-08 06:12:54
subprocess.py	text/x-script.python	7.56 KB	root/linksafe	0644	2026-01-08 06:12:54
taskgroups.py	text/x-script.python	9.33 KB	root/linksafe	0644	2026-01-08 06:12:55
tasks.py	text/plain	36.49 KB	root/linksafe	0644	2026-01-08 06:12:54
threads.py	text/plain	790 B	root/linksafe	0644	2026-01-08 06:12:55
timeouts.py	text/x-script.python	5.2 KB	root/linksafe	0644	2026-01-08 06:12:54
transports.py	text/plain	10.47 KB	root/linksafe	0644	2026-01-08 06:12:54
trsock.py	text/x-script.python	2.42 KB	root/linksafe	0644	2026-01-08 06:12:55
unix_events.py	text/plain	51.88 KB	root/linksafe	0644	2026-01-08 06:12:55
windows_events.py	text/plain	31.82 KB	root/linksafe	0644	2026-01-08 06:12:55
windows_utils.py	text/plain	4.94 KB	root/linksafe	0644	2026-01-08 06:12:55

Team Anon Force

https://t.me/Professor6T9x