Module `rsyscall.unistd.credentials`

Named for the excellent manpage about these process attributes, credentials(7)

Expand source code Browse git

"Named for the excellent manpage about these process attributes, credentials(7)"
from __future__ import annotations
import rsyscall.near as near
import typing as t

#### Classes ####
if t.TYPE_CHECKING:
    from rsyscall.handle.process import ChildProcess
import rsyscall.far

class CredentialsTask(rsyscall.far.Task):
    async def getuid(self) -> int:
        return (await _getuid(self.sysif))

    async def getgid(self) -> int:
        return (await _getgid(self.sysif))

    async def getpid(self) -> near.Process:
        return (await _getpid(self.sysif))

    async def getpgid(self) -> near.ProcessGroup:
        return (await _getpgid(self.sysif, None))

    async def setpgid(self, pgid: t.Optional[ChildProcess]=None) -> None:
        if pgid is None:
            await _setpgid(self.sysif, None, None)
        else:
            if pgid.task.pidns != self.pidns:
                raise rsyscall.far.NamespaceMismatchError(
                    "different pid namespaces", pgid.task.pidns, self.pidns)
            with pgid.borrow():
                await _setpgid(self.sysif, None, pgid._as_process_group())

    async def setsid(self) -> int:
        return (await _setsid(self.sysif))

#### Raw syscalls ####
from rsyscall.near.sysif import SyscallInterface
from rsyscall.sys.syscall import SYS

async def _getuid(sysif: SyscallInterface) -> int:
    return (await sysif.syscall(SYS.getuid))

async def _getgid(sysif: SyscallInterface) -> int:
    return (await sysif.syscall(SYS.getgid))

async def _getpid(sysif: SyscallInterface) -> near.Process:
    return near.Process(await sysif.syscall(SYS.getpid))

async def _getpgid(sysif: SyscallInterface, pid: t.Optional[near.Process]) -> near.ProcessGroup:
    if pid is None:
        pid = 0 # type: ignore
    return near.ProcessGroup(await sysif.syscall(SYS.getpgid, pid))

async def _setpgid(sysif: SyscallInterface,
                   pid: t.Optional[near.Process], pgid: t.Optional[near.ProcessGroup]) -> None:
    if pid is None:
        pid = 0 # type: ignore
    if pgid is None:
        pgid = 0 # type: ignore
    await sysif.syscall(SYS.setpgid, pid, pgid)

async def _setsid(sysif: SyscallInterface) -> int:
    return (await sysif.syscall(SYS.setsid))

Classes

class CredentialsTask (sysif: SyscallInterface, near_process: Process, fd_table: FDTable, address_space: AddressSpace, pidns: PidNamespace)

A wrapper around SyscallInterface which tracks the namespaces of the underlying process

Note that this is a base class for the more fully featured Task.

We store namespace objects to represent the namespaces that we believe that underlying processes is in. Since we have complete control over the process, we can make sure this belief is accurate, by updating our stored namespaces when the process changes namespace. That isn't done here; it's done in handle.Task.

Currently, we store only one PidNamespace. But each process actually has two pid namespaces:

the process's own pid namespace, which determines the pids returned from getpid, clone, and other syscalls.
the pid namespace that new children will be in.

The two pid namespaces only differ if we call unshare(CLONE.NEWPID). Currently we don't do that because unshare(CLONE.NEWPID) makes monitoring children more complex, since they can be deleted without leaving a zombie at any time if the pid namespace shuts down. But if we did call unshare(CLONE.NEWPID), we'd need to handle this right.

In the analogy to near and far pointers, this is like a segment register, if a segment register was write-only. Then we'd need to maintain the knowledge of what the segment register was set to, outside the segment register itself. That's what we do here.

There actually were systems where segment registers were, if not quite write-only, at least expensive to set and expensive to read. For example, x86_64 - the FS and GS segment registers can only be set via syscall. If you wanted to use segmentation on such systems, you'd probably have a structure much like this one.

Expand source code Browse git

class CredentialsTask(rsyscall.far.Task):
    async def getuid(self) -> int:
        return (await _getuid(self.sysif))

    async def getgid(self) -> int:
        return (await _getgid(self.sysif))

    async def getpid(self) -> near.Process:
        return (await _getpid(self.sysif))

    async def getpgid(self) -> near.ProcessGroup:
        return (await _getpgid(self.sysif, None))

    async def setpgid(self, pgid: t.Optional[ChildProcess]=None) -> None:
        if pgid is None:
            await _setpgid(self.sysif, None, None)
        else:
            if pgid.task.pidns != self.pidns:
                raise rsyscall.far.NamespaceMismatchError(
                    "different pid namespaces", pgid.task.pidns, self.pidns)
            with pgid.borrow():
                await _setpgid(self.sysif, None, pgid._as_process_group())

    async def setsid(self) -> int:
        return (await _setsid(self.sysif))

Ancestors

Task

Subclasses

Task

Class variables

var sysif : SyscallInterface
var near_process : Process
var fd_table : FDTable
var address_space : AddressSpace
var pidns : PidNamespace

Methods

async def getuid(self) ‑> int

Expand source code Browse git

async def getuid(self) -> int:
    return (await _getuid(self.sysif))

async def getgid(self) ‑> int

Expand source code Browse git

async def getgid(self) -> int:
    return (await _getgid(self.sysif))

async def getpid(self) ‑> Process

Expand source code Browse git

async def getpid(self) -> near.Process:
    return (await _getpid(self.sysif))

async def getpgid(self) ‑> ProcessGroup

Expand source code Browse git

async def getpgid(self) -> near.ProcessGroup:
    return (await _getpgid(self.sysif, None))

async def setpgid(self, pgid: t.Optional[ChildProcess] = None) ‑> None

Expand source code Browse git

async def setpgid(self, pgid: t.Optional[ChildProcess]=None) -> None:
    if pgid is None:
        await _setpgid(self.sysif, None, None)
    else:
        if pgid.task.pidns != self.pidns:
            raise rsyscall.far.NamespaceMismatchError(
                "different pid namespaces", pgid.task.pidns, self.pidns)
        with pgid.borrow():
            await _setpgid(self.sysif, None, pgid._as_process_group())

async def setsid(self) ‑> int

Expand source code Browse git

async def setsid(self) -> int:
    return (await _setsid(self.sysif))