diff --git a/README.md b/README.md index 67978ef..4dac0d1 100644 --- a/README.md +++ b/README.md @@ -257,4 +257,26 @@ For complete runnable Go programs: # Delete the ate-env namespace to remove all components: kubectl delete ns ate-env kubectl ate delete actor-template --atespace ate-env default-template -``` \ No newline at end of file +``` +## Guest ownership for snapshot restores + +`ate-env-guest --require-owner --workspace=/workspace` opts into Linux +controller fencing. Each gRPC call must carry `ate-owner-generation` (a +positive, monotonically increasing int64 issued outside the guest) and +`ate-owner-token`. A higher generation cancels and finishes the prior owner's +RPCs, drains tracked process groups, and admits the new owner. Lower +generations and a different token for the current generation are rejected. +This is fencing, not authentication: use an authenticated router and a trusted +controller; do not expose the guest directly to untrusted callers. + +`StartProcessRequest.process_id` accepts a caller-generated UUID. Killing an +unknown UUID records cancellation before a delayed start arrives. Duplicate +starts fail rather than replay a command. `KillAllProcesses` drains tracked +process groups before suspension. Output RPCs transfer at most 64 KiB per +message and drain the final output after the process reaper finishes. + +File reads, writes, and `RemovePath` use Go's rooted filesystem operations, +including symlink and rename protection. Ownership mode also holds a rooted +working-directory descriptor through Linux process startup. Processes are +still arbitrary programs inside the actor sandbox; filesystem RPC confinement +is not a substitute for the sandbox itself. diff --git a/clients/python/scripts/gen-protos.sh b/clients/python/scripts/gen-protos.sh index c846023..0d9f375 100755 --- a/clients/python/scripts/gen-protos.sh +++ b/clients/python/scripts/gen-protos.sh @@ -41,3 +41,9 @@ for f in "${OUT_DIR}"/ateenv/v1alpha/*_pb2_grpc.py; do sed -i.bak 's/^from ateenv\.v1alpha import \(.*\)$/from . import \1/' "${f}" rm -f "${f}.bak" done + +# Keep generated files subject to the repository's normal license gate. +for f in "${OUT_DIR}"/ateenv/v1alpha/*_pb2.py "${OUT_DIR}"/ateenv/v1alpha/*_pb2.pyi "${OUT_DIR}"/ateenv/v1alpha/*_pb2_grpc.py; do + { cat hack/boilerplate/sh.txt; cat "$f"; } > "${f}.tmp" + mv "${f}.tmp" "$f" +done diff --git a/clients/python/src/ate_env/_gen/ateenv/v1alpha/guest_pb2.py b/clients/python/src/ate_env/_gen/ateenv/v1alpha/guest_pb2.py index e0d6e8b..d5a4723 100644 --- a/clients/python/src/ate_env/_gen/ateenv/v1alpha/guest_pb2.py +++ b/clients/python/src/ate_env/_gen/ateenv/v1alpha/guest_pb2.py @@ -14,63 +14,76 @@ # -*- coding: utf-8 -*- # Generated by the protocol buffer compiler. DO NOT EDIT! +# NO CHECKED-IN PROTOBUF GENCODE # source: ateenv/v1alpha/guest.proto -# Protobuf Python Version: 4.25.1 +# Protobuf Python Version: 7.35.1 """Generated protocol buffer code.""" from google.protobuf import descriptor as _descriptor from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import runtime_version as _runtime_version from google.protobuf import symbol_database as _symbol_database from google.protobuf.internal import builder as _builder +_runtime_version.ValidateProtobufRuntimeVersion( + _runtime_version.Domain.PUBLIC, + 7, + 35, + 1, + '', + 'ateenv/v1alpha/guest.proto' +) # @@protoc_insertion_point(imports) _sym_db = _symbol_database.Default() from google.protobuf import timestamp_pb2 as google_dot_protobuf_dot_timestamp__pb2 +from google.protobuf import empty_pb2 as google_dot_protobuf_dot_empty__pb2 -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1a\x61teenv/v1alpha/guest.proto\x12\x0e\x61teenv.v1alpha\x1a\x1fgoogle/protobuf/timestamp.proto\"\xc0\x01\n\x07Process\x12\x12\n\nprocess_id\x18\x01 \x01(\t\x12-\n\x06status\x18\x02 \x01(\x0e\x32\x1d.ateenv.v1alpha.ProcessStatus\x12\x11\n\texit_code\x18\x03 \x01(\x05\x12.\n\nstarted_at\x18\x04 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12/\n\x0b\x66inished_at\x18\x05 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\"\x9a\x01\n\x13StartProcessRequest\x12\x0f\n\x07\x63ommand\x18\x01 \x03(\t\x12\x0b\n\x03\x63wd\x18\x02 \x01(\t\x12\x39\n\x03\x65nv\x18\x03 \x03(\x0b\x32,.ateenv.v1alpha.StartProcessRequest.EnvEntry\x1a*\n\x08\x45nvEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"*\n\x14StartProcessResponse\x12\x12\n\nprocess_id\x18\x01 \x01(\t\"\'\n\x11GetProcessRequest\x12\x12\n\nprocess_id\x18\x01 \x01(\t\"o\n\x1bStreamProcessOutputsRequest\x12\x12\n\nprocess_id\x18\x01 \x01(\t\x12\x15\n\rstdout_offset\x18\x02 \x01(\x03\x12\x15\n\rstderr_offset\x18\x03 \x01(\x03\x12\x0e\n\x06\x66ollow\x18\x04 \x01(\x08\"I\n\x0bOutputChunk\x12,\n\x06source\x18\x01 \x01(\x0e\x32\x1c.ateenv.v1alpha.OutputSource\x12\x0c\n\x04\x64\x61ta\x18\x02 \x01(\x0c\"(\n\x12KillProcessRequest\x12\x12\n\nprocess_id\x18\x01 \x01(\t\"(\n\x13KillProcessResponse\x12\x11\n\texit_code\x18\x01 \x01(\x05\"\x1f\n\x0fReadFileRequest\x12\x0c\n\x04path\x18\x01 \x01(\t\"\x19\n\tFileChunk\x12\x0c\n\x04\x64\x61ta\x18\x01 \x01(\x0c\"=\n\x10WriteFileRequest\x12\x0c\n\x04path\x18\x01 \x01(\t\x12\r\n\x05\x63hunk\x18\x02 \x01(\x0c\x12\x0c\n\x04mode\x18\x03 \x01(\r\"*\n\x11WriteFileResponse\x12\x15\n\rbytes_written\x18\x01 \x01(\x03*\xa3\x01\n\rProcessStatus\x12\x1e\n\x1aPROCESS_STATUS_UNSPECIFIED\x10\x00\x12\x1a\n\x16PROCESS_STATUS_RUNNING\x10\x01\x12\x1c\n\x18PROCESS_STATUS_COMPLETED\x10\x02\x12\x19\n\x15PROCESS_STATUS_FAILED\x10\x03\x12\x1d\n\x19PROCESS_STATUS_TERMINATED\x10\x04*a\n\x0cOutputSource\x12\x1d\n\x19OUTPUT_SOURCE_UNSPECIFIED\x10\x00\x12\x18\n\x14OUTPUT_SOURCE_STDOUT\x10\x01\x12\x18\n\x14OUTPUT_SOURCE_STDERR\x10\x02\x32\xf1\x02\n\x0eProcessService\x12Y\n\x0cStartProcess\x12#.ateenv.v1alpha.StartProcessRequest\x1a$.ateenv.v1alpha.StartProcessResponse\x12H\n\nGetProcess\x12!.ateenv.v1alpha.GetProcessRequest\x1a\x17.ateenv.v1alpha.Process\x12\x62\n\x14StreamProcessOutputs\x12+.ateenv.v1alpha.StreamProcessOutputsRequest\x1a\x1b.ateenv.v1alpha.OutputChunk0\x01\x12V\n\x0bKillProcess\x12\".ateenv.v1alpha.KillProcessRequest\x1a#.ateenv.v1alpha.KillProcessResponse2\xb1\x01\n\x11\x46ileSystemService\x12H\n\x08ReadFile\x12\x1f.ateenv.v1alpha.ReadFileRequest\x1a\x19.ateenv.v1alpha.FileChunk0\x01\x12R\n\tWriteFile\x12 .ateenv.v1alpha.WriteFileRequest\x1a!.ateenv.v1alpha.WriteFileResponse(\x01\x42\x43ZAgithub.com/agent-substrate/env/proto/ateenv/v1alpha;ateenvv1alphab\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1a\x61teenv/v1alpha/guest.proto\x12\x0e\x61teenv.v1alpha\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x1bgoogle/protobuf/empty.proto\"\xc0\x01\n\x07Process\x12\x12\n\nprocess_id\x18\x01 \x01(\t\x12-\n\x06status\x18\x02 \x01(\x0e\x32\x1d.ateenv.v1alpha.ProcessStatus\x12\x11\n\texit_code\x18\x03 \x01(\x05\x12.\n\nstarted_at\x18\x04 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12/\n\x0b\x66inished_at\x18\x05 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\"\xae\x01\n\x13StartProcessRequest\x12\x0f\n\x07\x63ommand\x18\x01 \x03(\t\x12\x0b\n\x03\x63wd\x18\x02 \x01(\t\x12\x39\n\x03\x65nv\x18\x03 \x03(\x0b\x32,.ateenv.v1alpha.StartProcessRequest.EnvEntry\x12\x12\n\nprocess_id\x18\x04 \x01(\t\x1a*\n\x08\x45nvEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"*\n\x14StartProcessResponse\x12\x12\n\nprocess_id\x18\x01 \x01(\t\"\'\n\x11GetProcessRequest\x12\x12\n\nprocess_id\x18\x01 \x01(\t\"o\n\x1bStreamProcessOutputsRequest\x12\x12\n\nprocess_id\x18\x01 \x01(\t\x12\x15\n\rstdout_offset\x18\x02 \x01(\x03\x12\x15\n\rstderr_offset\x18\x03 \x01(\x03\x12\x0e\n\x06\x66ollow\x18\x04 \x01(\x08\"I\n\x0bOutputChunk\x12,\n\x06source\x18\x01 \x01(\x0e\x32\x1c.ateenv.v1alpha.OutputSource\x12\x0c\n\x04\x64\x61ta\x18\x02 \x01(\x0c\"(\n\x12KillProcessRequest\x12\x12\n\nprocess_id\x18\x01 \x01(\t\"(\n\x13KillProcessResponse\x12\x11\n\texit_code\x18\x01 \x01(\x05\"\x1f\n\x0fReadFileRequest\x12\x0c\n\x04path\x18\x01 \x01(\t\"\x19\n\tFileChunk\x12\x0c\n\x04\x64\x61ta\x18\x01 \x01(\x0c\"=\n\x10WriteFileRequest\x12\x0c\n\x04path\x18\x01 \x01(\t\x12\r\n\x05\x63hunk\x18\x02 \x01(\x0c\x12\x0c\n\x04mode\x18\x03 \x01(\r\"*\n\x11WriteFileResponse\x12\x15\n\rbytes_written\x18\x01 \x01(\x03\"C\n\x11RemovePathRequest\x12\x0c\n\x04path\x18\x01 \x01(\t\x12\x11\n\trecursive\x18\x02 \x01(\x08\x12\r\n\x05\x66orce\x18\x03 \x01(\x08*\xa3\x01\n\rProcessStatus\x12\x1e\n\x1aPROCESS_STATUS_UNSPECIFIED\x10\x00\x12\x1a\n\x16PROCESS_STATUS_RUNNING\x10\x01\x12\x1c\n\x18PROCESS_STATUS_COMPLETED\x10\x02\x12\x19\n\x15PROCESS_STATUS_FAILED\x10\x03\x12\x1d\n\x19PROCESS_STATUS_TERMINATED\x10\x04*a\n\x0cOutputSource\x12\x1d\n\x19OUTPUT_SOURCE_UNSPECIFIED\x10\x00\x12\x18\n\x14OUTPUT_SOURCE_STDOUT\x10\x01\x12\x18\n\x14OUTPUT_SOURCE_STDERR\x10\x02\x32\xb5\x03\n\x0eProcessService\x12Y\n\x0cStartProcess\x12#.ateenv.v1alpha.StartProcessRequest\x1a$.ateenv.v1alpha.StartProcessResponse\x12H\n\nGetProcess\x12!.ateenv.v1alpha.GetProcessRequest\x1a\x17.ateenv.v1alpha.Process\x12\x62\n\x14StreamProcessOutputs\x12+.ateenv.v1alpha.StreamProcessOutputsRequest\x1a\x1b.ateenv.v1alpha.OutputChunk0\x01\x12V\n\x0bKillProcess\x12\".ateenv.v1alpha.KillProcessRequest\x1a#.ateenv.v1alpha.KillProcessResponse\x12\x42\n\x10KillAllProcesses\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.Empty2\xfa\x01\n\x11\x46ileSystemService\x12H\n\x08ReadFile\x12\x1f.ateenv.v1alpha.ReadFileRequest\x1a\x19.ateenv.v1alpha.FileChunk0\x01\x12R\n\tWriteFile\x12 .ateenv.v1alpha.WriteFileRequest\x1a!.ateenv.v1alpha.WriteFileResponse(\x01\x12G\n\nRemovePath\x12!.ateenv.v1alpha.RemovePathRequest\x1a\x16.google.protobuf.EmptyBCZAgithub.com/agent-substrate/env/proto/ateenv/v1alpha;ateenvv1alphab\x06proto3') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) _builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'ateenv.v1alpha.guest_pb2', _globals) -if _descriptor._USE_C_DESCRIPTORS == False: - _globals['DESCRIPTOR']._options = None +if not _descriptor._USE_C_DESCRIPTORS: + _globals['DESCRIPTOR']._loaded_options = None _globals['DESCRIPTOR']._serialized_options = b'ZAgithub.com/agent-substrate/env/proto/ateenv/v1alpha;ateenvv1alpha' - _globals['_STARTPROCESSREQUEST_ENVENTRY']._options = None + _globals['_STARTPROCESSREQUEST_ENVENTRY']._loaded_options = None _globals['_STARTPROCESSREQUEST_ENVENTRY']._serialized_options = b'8\001' - _globals['_PROCESSSTATUS']._serialized_start=956 - _globals['_PROCESSSTATUS']._serialized_end=1119 - _globals['_OUTPUTSOURCE']._serialized_start=1121 - _globals['_OUTPUTSOURCE']._serialized_end=1218 - _globals['_PROCESS']._serialized_start=80 - _globals['_PROCESS']._serialized_end=272 - _globals['_STARTPROCESSREQUEST']._serialized_start=275 - _globals['_STARTPROCESSREQUEST']._serialized_end=429 - _globals['_STARTPROCESSREQUEST_ENVENTRY']._serialized_start=387 - _globals['_STARTPROCESSREQUEST_ENVENTRY']._serialized_end=429 - _globals['_STARTPROCESSRESPONSE']._serialized_start=431 - _globals['_STARTPROCESSRESPONSE']._serialized_end=473 - _globals['_GETPROCESSREQUEST']._serialized_start=475 - _globals['_GETPROCESSREQUEST']._serialized_end=514 - _globals['_STREAMPROCESSOUTPUTSREQUEST']._serialized_start=516 - _globals['_STREAMPROCESSOUTPUTSREQUEST']._serialized_end=627 - _globals['_OUTPUTCHUNK']._serialized_start=629 - _globals['_OUTPUTCHUNK']._serialized_end=702 - _globals['_KILLPROCESSREQUEST']._serialized_start=704 - _globals['_KILLPROCESSREQUEST']._serialized_end=744 - _globals['_KILLPROCESSRESPONSE']._serialized_start=746 - _globals['_KILLPROCESSRESPONSE']._serialized_end=786 - _globals['_READFILEREQUEST']._serialized_start=788 - _globals['_READFILEREQUEST']._serialized_end=819 - _globals['_FILECHUNK']._serialized_start=821 - _globals['_FILECHUNK']._serialized_end=846 - _globals['_WRITEFILEREQUEST']._serialized_start=848 - _globals['_WRITEFILEREQUEST']._serialized_end=909 - _globals['_WRITEFILERESPONSE']._serialized_start=911 - _globals['_WRITEFILERESPONSE']._serialized_end=953 - _globals['_PROCESSSERVICE']._serialized_start=1221 - _globals['_PROCESSSERVICE']._serialized_end=1590 - _globals['_FILESYSTEMSERVICE']._serialized_start=1593 - _globals['_FILESYSTEMSERVICE']._serialized_end=1770 + _globals['_PROCESSSTATUS']._serialized_start=1074 + _globals['_PROCESSSTATUS']._serialized_end=1237 + _globals['_OUTPUTSOURCE']._serialized_start=1239 + _globals['_OUTPUTSOURCE']._serialized_end=1336 + _globals['_PROCESS']._serialized_start=109 + _globals['_PROCESS']._serialized_end=301 + _globals['_STARTPROCESSREQUEST']._serialized_start=304 + _globals['_STARTPROCESSREQUEST']._serialized_end=478 + _globals['_STARTPROCESSREQUEST_ENVENTRY']._serialized_start=436 + _globals['_STARTPROCESSREQUEST_ENVENTRY']._serialized_end=478 + _globals['_STARTPROCESSRESPONSE']._serialized_start=480 + _globals['_STARTPROCESSRESPONSE']._serialized_end=522 + _globals['_GETPROCESSREQUEST']._serialized_start=524 + _globals['_GETPROCESSREQUEST']._serialized_end=563 + _globals['_STREAMPROCESSOUTPUTSREQUEST']._serialized_start=565 + _globals['_STREAMPROCESSOUTPUTSREQUEST']._serialized_end=676 + _globals['_OUTPUTCHUNK']._serialized_start=678 + _globals['_OUTPUTCHUNK']._serialized_end=751 + _globals['_KILLPROCESSREQUEST']._serialized_start=753 + _globals['_KILLPROCESSREQUEST']._serialized_end=793 + _globals['_KILLPROCESSRESPONSE']._serialized_start=795 + _globals['_KILLPROCESSRESPONSE']._serialized_end=835 + _globals['_READFILEREQUEST']._serialized_start=837 + _globals['_READFILEREQUEST']._serialized_end=868 + _globals['_FILECHUNK']._serialized_start=870 + _globals['_FILECHUNK']._serialized_end=895 + _globals['_WRITEFILEREQUEST']._serialized_start=897 + _globals['_WRITEFILEREQUEST']._serialized_end=958 + _globals['_WRITEFILERESPONSE']._serialized_start=960 + _globals['_WRITEFILERESPONSE']._serialized_end=1002 + _globals['_REMOVEPATHREQUEST']._serialized_start=1004 + _globals['_REMOVEPATHREQUEST']._serialized_end=1071 + _globals['_PROCESSSERVICE']._serialized_start=1339 + _globals['_PROCESSSERVICE']._serialized_end=1776 + _globals['_FILESYSTEMSERVICE']._serialized_start=1779 + _globals['_FILESYSTEMSERVICE']._serialized_end=2029 # @@protoc_insertion_point(module_scope) diff --git a/clients/python/src/ate_env/_gen/ateenv/v1alpha/guest_pb2.pyi b/clients/python/src/ate_env/_gen/ateenv/v1alpha/guest_pb2.pyi index 51fbd72..b7870ec 100644 --- a/clients/python/src/ate_env/_gen/ateenv/v1alpha/guest_pb2.pyi +++ b/clients/python/src/ate_env/_gen/ateenv/v1alpha/guest_pb2.pyi @@ -1,23 +1,27 @@ -// Copyright 2026 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime from google.protobuf import timestamp_pb2 as _timestamp_pb2 +from google.protobuf import empty_pb2 as _empty_pb2 from google.protobuf.internal import containers as _containers from google.protobuf.internal import enum_type_wrapper as _enum_type_wrapper from google.protobuf import descriptor as _descriptor from google.protobuf import message as _message -from typing import ClassVar as _ClassVar, Iterable as _Iterable, Mapping as _Mapping, Optional as _Optional, Union as _Union +from collections.abc import Iterable as _Iterable, Mapping as _Mapping +from typing import ClassVar as _ClassVar, Optional as _Optional, Union as _Union DESCRIPTOR: _descriptor.FileDescriptor @@ -55,10 +59,10 @@ class Process(_message.Message): exit_code: int started_at: _timestamp_pb2.Timestamp finished_at: _timestamp_pb2.Timestamp - def __init__(self, process_id: _Optional[str] = ..., status: _Optional[_Union[ProcessStatus, str]] = ..., exit_code: _Optional[int] = ..., started_at: _Optional[_Union[_timestamp_pb2.Timestamp, _Mapping]] = ..., finished_at: _Optional[_Union[_timestamp_pb2.Timestamp, _Mapping]] = ...) -> None: ... + def __init__(self, process_id: _Optional[str] = ..., status: _Optional[_Union[ProcessStatus, str]] = ..., exit_code: _Optional[int] = ..., started_at: _Optional[_Union[datetime.datetime, _timestamp_pb2.Timestamp, _Mapping]] = ..., finished_at: _Optional[_Union[datetime.datetime, _timestamp_pb2.Timestamp, _Mapping]] = ...) -> None: ... class StartProcessRequest(_message.Message): - __slots__ = ("command", "cwd", "env") + __slots__ = ("command", "cwd", "env", "process_id") class EnvEntry(_message.Message): __slots__ = ("key", "value") KEY_FIELD_NUMBER: _ClassVar[int] @@ -69,10 +73,12 @@ class StartProcessRequest(_message.Message): COMMAND_FIELD_NUMBER: _ClassVar[int] CWD_FIELD_NUMBER: _ClassVar[int] ENV_FIELD_NUMBER: _ClassVar[int] + PROCESS_ID_FIELD_NUMBER: _ClassVar[int] command: _containers.RepeatedScalarFieldContainer[str] cwd: str env: _containers.ScalarMap[str, str] - def __init__(self, command: _Optional[_Iterable[str]] = ..., cwd: _Optional[str] = ..., env: _Optional[_Mapping[str, str]] = ...) -> None: ... + process_id: str + def __init__(self, command: _Optional[_Iterable[str]] = ..., cwd: _Optional[str] = ..., env: _Optional[_Mapping[str, str]] = ..., process_id: _Optional[str] = ...) -> None: ... class StartProcessResponse(_message.Message): __slots__ = ("process_id",) @@ -96,7 +102,7 @@ class StreamProcessOutputsRequest(_message.Message): stdout_offset: int stderr_offset: int follow: bool - def __init__(self, process_id: _Optional[str] = ..., stdout_offset: _Optional[int] = ..., stderr_offset: _Optional[int] = ..., follow: bool = ...) -> None: ... + def __init__(self, process_id: _Optional[str] = ..., stdout_offset: _Optional[int] = ..., stderr_offset: _Optional[int] = ..., follow: _Optional[bool] = ...) -> None: ... class OutputChunk(_message.Message): __slots__ = ("source", "data") @@ -145,3 +151,13 @@ class WriteFileResponse(_message.Message): BYTES_WRITTEN_FIELD_NUMBER: _ClassVar[int] bytes_written: int def __init__(self, bytes_written: _Optional[int] = ...) -> None: ... + +class RemovePathRequest(_message.Message): + __slots__ = ("path", "recursive", "force") + PATH_FIELD_NUMBER: _ClassVar[int] + RECURSIVE_FIELD_NUMBER: _ClassVar[int] + FORCE_FIELD_NUMBER: _ClassVar[int] + path: str + recursive: bool + force: bool + def __init__(self, path: _Optional[str] = ..., recursive: _Optional[bool] = ..., force: _Optional[bool] = ...) -> None: ... diff --git a/clients/python/src/ate_env/_gen/ateenv/v1alpha/guest_pb2_grpc.py b/clients/python/src/ate_env/_gen/ateenv/v1alpha/guest_pb2_grpc.py index 6a2d212..ada4a54 100644 --- a/clients/python/src/ate_env/_gen/ateenv/v1alpha/guest_pb2_grpc.py +++ b/clients/python/src/ate_env/_gen/ateenv/v1alpha/guest_pb2_grpc.py @@ -15,11 +15,32 @@ # Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! """Client and server classes corresponding to protobuf-defined services.""" import grpc +import warnings from . import guest_pb2 as ateenv_dot_v1alpha_dot_guest__pb2 +from google.protobuf import empty_pb2 as google_dot_protobuf_dot_empty__pb2 +GRPC_GENERATED_VERSION = '1.83.0' +GRPC_VERSION = grpc.__version__ +_version_not_supported = False -class ProcessServiceStub(object): +try: + from grpc._utilities import first_version_is_lower + _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION) +except ImportError: + _version_not_supported = True + +if _version_not_supported: + raise RuntimeError( + f'The grpc package installed is at version {GRPC_VERSION},' + + ' but the generated code in ateenv/v1alpha/guest_pb2_grpc.py depends on' + + f' grpcio>={GRPC_GENERATED_VERSION}.' + + f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}' + + f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.' + ) + + +class ProcessServiceStub: """============================================================================ --- SERVICES --- ============================================================================ @@ -38,25 +59,30 @@ def __init__(self, channel): '/ateenv.v1alpha.ProcessService/StartProcess', request_serializer=ateenv_dot_v1alpha_dot_guest__pb2.StartProcessRequest.SerializeToString, response_deserializer=ateenv_dot_v1alpha_dot_guest__pb2.StartProcessResponse.FromString, - ) + _registered_method=True) self.GetProcess = channel.unary_unary( '/ateenv.v1alpha.ProcessService/GetProcess', request_serializer=ateenv_dot_v1alpha_dot_guest__pb2.GetProcessRequest.SerializeToString, response_deserializer=ateenv_dot_v1alpha_dot_guest__pb2.Process.FromString, - ) + _registered_method=True) self.StreamProcessOutputs = channel.unary_stream( '/ateenv.v1alpha.ProcessService/StreamProcessOutputs', request_serializer=ateenv_dot_v1alpha_dot_guest__pb2.StreamProcessOutputsRequest.SerializeToString, response_deserializer=ateenv_dot_v1alpha_dot_guest__pb2.OutputChunk.FromString, - ) + _registered_method=True) self.KillProcess = channel.unary_unary( '/ateenv.v1alpha.ProcessService/KillProcess', request_serializer=ateenv_dot_v1alpha_dot_guest__pb2.KillProcessRequest.SerializeToString, response_deserializer=ateenv_dot_v1alpha_dot_guest__pb2.KillProcessResponse.FromString, - ) + _registered_method=True) + self.KillAllProcesses = channel.unary_unary( + '/ateenv.v1alpha.ProcessService/KillAllProcesses', + request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + _registered_method=True) -class ProcessServiceServicer(object): +class ProcessServiceServicer: """============================================================================ --- SERVICES --- ============================================================================ @@ -94,6 +120,13 @@ def KillProcess(self, request, context): context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') + def KillAllProcesses(self, request, context): + """Drain tracked processes before suspending an environment. + """ + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + def add_ProcessServiceServicer_to_server(servicer, server): rpc_method_handlers = { @@ -117,14 +150,20 @@ def add_ProcessServiceServicer_to_server(servicer, server): request_deserializer=ateenv_dot_v1alpha_dot_guest__pb2.KillProcessRequest.FromString, response_serializer=ateenv_dot_v1alpha_dot_guest__pb2.KillProcessResponse.SerializeToString, ), + 'KillAllProcesses': grpc.unary_unary_rpc_method_handler( + servicer.KillAllProcesses, + request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + ), } generic_handler = grpc.method_handlers_generic_handler( 'ateenv.v1alpha.ProcessService', rpc_method_handlers) server.add_generic_rpc_handlers((generic_handler,)) + server.add_registered_method_handlers('ateenv.v1alpha.ProcessService', rpc_method_handlers) # This class is part of an EXPERIMENTAL API. -class ProcessService(object): +class ProcessService: """============================================================================ --- SERVICES --- ============================================================================ @@ -144,11 +183,21 @@ def StartProcess(request, wait_for_ready=None, timeout=None, metadata=None): - return grpc.experimental.unary_unary(request, target, '/ateenv.v1alpha.ProcessService/StartProcess', + return grpc.experimental.unary_unary( + request, + target, + '/ateenv.v1alpha.ProcessService/StartProcess', ateenv_dot_v1alpha_dot_guest__pb2.StartProcessRequest.SerializeToString, ateenv_dot_v1alpha_dot_guest__pb2.StartProcessResponse.FromString, - options, channel_credentials, - insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) @staticmethod def GetProcess(request, @@ -161,11 +210,21 @@ def GetProcess(request, wait_for_ready=None, timeout=None, metadata=None): - return grpc.experimental.unary_unary(request, target, '/ateenv.v1alpha.ProcessService/GetProcess', + return grpc.experimental.unary_unary( + request, + target, + '/ateenv.v1alpha.ProcessService/GetProcess', ateenv_dot_v1alpha_dot_guest__pb2.GetProcessRequest.SerializeToString, ateenv_dot_v1alpha_dot_guest__pb2.Process.FromString, - options, channel_credentials, - insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) @staticmethod def StreamProcessOutputs(request, @@ -178,11 +237,21 @@ def StreamProcessOutputs(request, wait_for_ready=None, timeout=None, metadata=None): - return grpc.experimental.unary_stream(request, target, '/ateenv.v1alpha.ProcessService/StreamProcessOutputs', + return grpc.experimental.unary_stream( + request, + target, + '/ateenv.v1alpha.ProcessService/StreamProcessOutputs', ateenv_dot_v1alpha_dot_guest__pb2.StreamProcessOutputsRequest.SerializeToString, ateenv_dot_v1alpha_dot_guest__pb2.OutputChunk.FromString, - options, channel_credentials, - insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) @staticmethod def KillProcess(request, @@ -195,14 +264,51 @@ def KillProcess(request, wait_for_ready=None, timeout=None, metadata=None): - return grpc.experimental.unary_unary(request, target, '/ateenv.v1alpha.ProcessService/KillProcess', + return grpc.experimental.unary_unary( + request, + target, + '/ateenv.v1alpha.ProcessService/KillProcess', ateenv_dot_v1alpha_dot_guest__pb2.KillProcessRequest.SerializeToString, ateenv_dot_v1alpha_dot_guest__pb2.KillProcessResponse.FromString, - options, channel_credentials, - insecure, call_credentials, compression, wait_for_ready, timeout, metadata) - + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) -class FileSystemServiceStub(object): + @staticmethod + def KillAllProcesses(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/ateenv.v1alpha.ProcessService/KillAllProcesses', + google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + google_dot_protobuf_dot_empty__pb2.Empty.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) + + +class FileSystemServiceStub: """FileSystemService provides streaming file reading and writing capabilities inside the container rootfs/workspace to prevent memory exhaustion (OOM). """ @@ -217,15 +323,20 @@ def __init__(self, channel): '/ateenv.v1alpha.FileSystemService/ReadFile', request_serializer=ateenv_dot_v1alpha_dot_guest__pb2.ReadFileRequest.SerializeToString, response_deserializer=ateenv_dot_v1alpha_dot_guest__pb2.FileChunk.FromString, - ) + _registered_method=True) self.WriteFile = channel.stream_unary( '/ateenv.v1alpha.FileSystemService/WriteFile', request_serializer=ateenv_dot_v1alpha_dot_guest__pb2.WriteFileRequest.SerializeToString, response_deserializer=ateenv_dot_v1alpha_dot_guest__pb2.WriteFileResponse.FromString, - ) + _registered_method=True) + self.RemovePath = channel.unary_unary( + '/ateenv.v1alpha.FileSystemService/RemovePath', + request_serializer=ateenv_dot_v1alpha_dot_guest__pb2.RemovePathRequest.SerializeToString, + response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + _registered_method=True) -class FileSystemServiceServicer(object): +class FileSystemServiceServicer: """FileSystemService provides streaming file reading and writing capabilities inside the container rootfs/workspace to prevent memory exhaustion (OOM). """ @@ -244,6 +355,13 @@ def WriteFile(self, request_iterator, context): context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') + def RemovePath(self, request, context): + """Remove a file or directory within the configured filesystem root. + """ + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + def add_FileSystemServiceServicer_to_server(servicer, server): rpc_method_handlers = { @@ -257,14 +375,20 @@ def add_FileSystemServiceServicer_to_server(servicer, server): request_deserializer=ateenv_dot_v1alpha_dot_guest__pb2.WriteFileRequest.FromString, response_serializer=ateenv_dot_v1alpha_dot_guest__pb2.WriteFileResponse.SerializeToString, ), + 'RemovePath': grpc.unary_unary_rpc_method_handler( + servicer.RemovePath, + request_deserializer=ateenv_dot_v1alpha_dot_guest__pb2.RemovePathRequest.FromString, + response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + ), } generic_handler = grpc.method_handlers_generic_handler( 'ateenv.v1alpha.FileSystemService', rpc_method_handlers) server.add_generic_rpc_handlers((generic_handler,)) + server.add_registered_method_handlers('ateenv.v1alpha.FileSystemService', rpc_method_handlers) # This class is part of an EXPERIMENTAL API. -class FileSystemService(object): +class FileSystemService: """FileSystemService provides streaming file reading and writing capabilities inside the container rootfs/workspace to prevent memory exhaustion (OOM). """ @@ -280,11 +404,21 @@ def ReadFile(request, wait_for_ready=None, timeout=None, metadata=None): - return grpc.experimental.unary_stream(request, target, '/ateenv.v1alpha.FileSystemService/ReadFile', + return grpc.experimental.unary_stream( + request, + target, + '/ateenv.v1alpha.FileSystemService/ReadFile', ateenv_dot_v1alpha_dot_guest__pb2.ReadFileRequest.SerializeToString, ateenv_dot_v1alpha_dot_guest__pb2.FileChunk.FromString, - options, channel_credentials, - insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) @staticmethod def WriteFile(request_iterator, @@ -297,8 +431,45 @@ def WriteFile(request_iterator, wait_for_ready=None, timeout=None, metadata=None): - return grpc.experimental.stream_unary(request_iterator, target, '/ateenv.v1alpha.FileSystemService/WriteFile', + return grpc.experimental.stream_unary( + request_iterator, + target, + '/ateenv.v1alpha.FileSystemService/WriteFile', ateenv_dot_v1alpha_dot_guest__pb2.WriteFileRequest.SerializeToString, ateenv_dot_v1alpha_dot_guest__pb2.WriteFileResponse.FromString, - options, channel_credentials, - insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) + + @staticmethod + def RemovePath(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/ateenv.v1alpha.FileSystemService/RemovePath', + ateenv_dot_v1alpha_dot_guest__pb2.RemovePathRequest.SerializeToString, + google_dot_protobuf_dot_empty__pb2.Empty.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) diff --git a/cmd/ate-env-guest/main.go b/cmd/ate-env-guest/main.go index d6f952f..46ef932 100644 --- a/cmd/ate-env-guest/main.go +++ b/cmd/ate-env-guest/main.go @@ -35,6 +35,7 @@ import ( func main() { listen := flag.String("listen", ":80", "address to serve the guest API on") logDir := flag.String("log-dir", "", "directory for process logs (defaults to /var/log/ate-jobs or temporary dir)") + requireOwner := flag.Bool("require-owner", false, "require trusted controller ownership metadata on every guest RPC") workspace := flag.String("workspace", "/", "workspace root directory") flag.Parse() @@ -47,6 +48,7 @@ func main() { ListenAddr: addr, LogDir: *logDir, Workspace: *workspace, + RequireOwner: *requireOwner, EnableProcess: true, EnableFileSystem: true, } diff --git a/guest/filesystem/service.go b/guest/filesystem/service.go index 5162969..ed67959 100644 --- a/guest/filesystem/service.go +++ b/guest/filesystem/service.go @@ -15,15 +15,18 @@ package filesystem import ( + "context" "errors" "io" "os" "path/filepath" "strings" + "syscall" ateenvv1alpha "github.com/agent-substrate/env/proto/ateenv/v1alpha" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/emptypb" ) const ( @@ -96,7 +99,11 @@ func (s *Service) resolveAndValidatePath(reqPath string) (string, error) { } else if s.rootDir != "" { targetPath = filepath.Clean(filepath.Join(s.rootDir, reqPath)) } else { - targetPath = filepath.Clean(reqPath) + var err error + targetPath, err = filepath.Abs(reqPath) + if err != nil { + return "", status.Error(codes.Internal, "resolving working directory") + } } // Boundary check if rootDirectory confinement is enabled @@ -119,7 +126,19 @@ func (s *Service) ReadFile(req *ateenvv1alpha.ReadFileRequest, stream ateenvv1al return err } - f, err := os.Open(filePath) + root, err := os.OpenRoot(s.rootDirectory()) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return status.Error(codes.NotFound, "file not found") + } + return status.Error(codes.Internal, "opening filesystem root") + } + defer root.Close() + relative, err := filepath.Rel(s.rootDirectory(), filePath) + if err != nil { + return status.Error(codes.InvalidArgument, "invalid file path") + } + f, err := root.OpenFile(relative, os.O_RDONLY|syscall.O_NONBLOCK, 0) if err != nil { if errors.Is(err, os.ErrNotExist) { return status.Errorf(codes.NotFound, "file %q not found", req.GetPath()) @@ -130,6 +149,9 @@ func (s *Service) ReadFile(req *ateenvv1alpha.ReadFileRequest, stream ateenvv1al return status.Errorf(codes.Internal, "failed to open file %q: %v", req.GetPath(), err) } defer f.Close() + if info, err := f.Stat(); err != nil || !info.Mode().IsRegular() { + return status.Error(codes.InvalidArgument, "path is not a regular file") + } buf := make([]byte, s.readBufferSize) for { @@ -158,11 +180,15 @@ func (s *Service) WriteFile(stream ateenvv1alpha.FileSystemService_WriteFileServ var totalBytes int64 var filePath string var reqPath string + var root *os.Root defer func() { if f != nil { _ = f.Close() } + if root != nil { + _ = root.Close() + } }() for { @@ -195,9 +221,20 @@ func (s *Service) WriteFile(stream ateenvv1alpha.FileSystemService_WriteFileServ filePath = validatedPath // Ensure parent directory exists - dir := filepath.Dir(filePath) - if dir != "" && dir != "." { - if err := os.MkdirAll(dir, 0755); err != nil { + if err := os.MkdirAll(s.rootDirectory(), 0755); err != nil { + return status.Error(codes.Internal, "creating filesystem root") + } + root, err = os.OpenRoot(s.rootDirectory()) + if err != nil { + return status.Error(codes.Internal, "opening filesystem root") + } + relative, err := filepath.Rel(s.rootDirectory(), filePath) + if err != nil { + return status.Error(codes.InvalidArgument, "invalid file path") + } + dir := filepath.Dir(relative) + if dir != "." { + if err := root.MkdirAll(dir, 0755); err != nil { return status.Errorf(codes.Internal, "failed to create parent directories for %q: %v", reqPath, err) } } @@ -207,13 +244,19 @@ func (s *Service) WriteFile(stream ateenvv1alpha.FileSystemService_WriteFileServ mode = os.FileMode(req.GetMode()) } - f, err = os.OpenFile(filePath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, mode) + f, err = root.OpenFile(relative, os.O_CREATE|os.O_WRONLY|syscall.O_NONBLOCK, mode) if err != nil { if errors.Is(err, os.ErrPermission) { return status.Errorf(codes.PermissionDenied, "permission denied opening %q: %v", reqPath, err) } return status.Errorf(codes.Internal, "failed to create file %q: %v", reqPath, err) } + if info, err := f.Stat(); err != nil || !info.Mode().IsRegular() { + return status.Error(codes.InvalidArgument, "path is not a regular file") + } + if err := f.Truncate(0); err != nil { + return status.Error(codes.Internal, "truncating file") + } } // Write chunk data @@ -227,3 +270,48 @@ func (s *Service) WriteFile(stream ateenvv1alpha.FileSystemService_WriteFileServ } } } + +func (s *Service) rootDirectory() string { + if s.rootDir != "" { + return s.rootDir + } + return "/" +} + +func (s *Service) RemovePath(ctx context.Context, req *ateenvv1alpha.RemovePathRequest) (*emptypb.Empty, error) { + if err := ctx.Err(); err != nil { + return nil, status.FromContextError(err).Err() + } + target, err := s.resolveAndValidatePath(req.GetPath()) + if err != nil { + return nil, err + } + relative, err := filepath.Rel(s.rootDirectory(), target) + if err != nil || relative == "." { + return nil, status.Error(codes.PermissionDenied, "cannot remove filesystem root") + } + root, err := os.OpenRoot(s.rootDirectory()) + if err != nil { + return nil, status.Error(codes.Internal, "opening filesystem root") + } + defer root.Close() + // RemoveAll treats a missing path as success; preserve the caller's force choice. + if _, err := root.Lstat(relative); err != nil { + if errors.Is(err, os.ErrNotExist) { + if req.GetForce() { + return &emptypb.Empty{}, nil + } + return nil, status.Error(codes.NotFound, "path not found") + } + return nil, status.Error(codes.PermissionDenied, "path unavailable within filesystem root") + } + if req.GetRecursive() { + err = root.RemoveAll(relative) + } else { + err = root.Remove(relative) + } + if err != nil { + return nil, status.Errorf(codes.FailedPrecondition, "removing path: %v", err) + } + return &emptypb.Empty{}, nil +} diff --git a/guest/filesystem/service_test.go b/guest/filesystem/service_test.go index bc78615..fff2601 100644 --- a/guest/filesystem/service_test.go +++ b/guest/filesystem/service_test.go @@ -349,3 +349,89 @@ func TestWriteFileMissingPath(t *testing.T) { t.Fatalf("expected InvalidArgument for missing path, got %v", err) } } + +func TestRootedFilesRejectSymlinkEscape(t *testing.T) { + root, outside := t.TempDir(), t.TempDir() + if err := os.WriteFile(filepath.Join(outside, "secret"), []byte("outside"), 0600); err != nil { + t.Fatal(err) + } + if err := os.Symlink(outside, filepath.Join(root, "escape")); err != nil { + t.Fatal(err) + } + client, cleanup := setupTestFileSystemServer(t, Config{RootDirectory: root}) + defer cleanup() + ctx := context.Background() + reader, err := client.ReadFile(ctx, &ateenvv1alpha.ReadFileRequest{Path: "escape/secret"}) + if err == nil { + _, err = reader.Recv() + } + if err == nil { + t.Fatal("symlink read escaped root") + } + writer, err := client.WriteFile(ctx) + if err != nil { + t.Fatal(err) + } + _ = writer.Send(&ateenvv1alpha.WriteFileRequest{Path: "escape/secret", Chunk: []byte("changed")}) + if _, err = writer.CloseAndRecv(); err == nil { + t.Fatal("symlink write escaped root") + } + contents, err := os.ReadFile(filepath.Join(outside, "secret")) + if err != nil || string(contents) != "outside" { + t.Fatal("outside file was changed") + } +} + +func TestRootedRemove(t *testing.T) { + root, outside := t.TempDir(), t.TempDir() + client, cleanup := setupTestFileSystemServer(t, Config{RootDirectory: root}) + defer cleanup() + if err := os.WriteFile(filepath.Join(outside, "keep"), []byte("outside"), 0600); err != nil { + t.Fatal(err) + } + if err := os.Symlink(outside, filepath.Join(root, "escape")); err != nil { + t.Fatal(err) + } + if _, err := client.RemovePath(t.Context(), &ateenvv1alpha.RemovePathRequest{Path: "escape/keep", Recursive: true}); err == nil { + t.Fatal("removed outside path") + } + if _, err := client.RemovePath(t.Context(), &ateenvv1alpha.RemovePathRequest{Path: "escape", Recursive: true}); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(filepath.Join(outside, "keep")); err != nil { + t.Fatal("outside file changed", err) + } + if _, err := client.RemovePath(t.Context(), &ateenvv1alpha.RemovePathRequest{Path: ".", Recursive: true}); status.Code(err) != codes.PermissionDenied { + t.Fatal(err) + } + if _, err := client.RemovePath(t.Context(), &ateenvv1alpha.RemovePathRequest{Path: "missing"}); status.Code(err) != codes.NotFound { + t.Fatal(err) + } + if _, err := client.RemovePath(t.Context(), &ateenvv1alpha.RemovePathRequest{Path: "missing", Force: true}); err != nil { + t.Fatal(err) + } +} + +func TestUnconfinedRelativePaths(t *testing.T) { + t.Chdir(t.TempDir()) + client, cleanup := setupTestFileSystemServer(t, Config{}) + defer cleanup() + stream, err := client.WriteFile(t.Context()) + if err != nil { + t.Fatal(err) + } + if err := stream.Send(&ateenvv1alpha.WriteFileRequest{Path: "relative.txt", Chunk: []byte("relative")}); err != nil { + t.Fatal(err) + } + if _, err := stream.CloseAndRecv(); err != nil { + t.Fatal(err) + } + read, err := client.ReadFile(t.Context(), &ateenvv1alpha.ReadFileRequest{Path: "relative.txt"}) + if err != nil { + t.Fatal(err) + } + chunk, err := read.Recv() + if err != nil || string(chunk.GetData()) != "relative" { + t.Fatal(chunk, err) + } +} diff --git a/guest/ownership.go b/guest/ownership.go new file mode 100644 index 0000000..0cc91a1 --- /dev/null +++ b/guest/ownership.go @@ -0,0 +1,131 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package guest + +import ( + "context" + "strconv" + "sync" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" +) + +// Ownership metadata is issued by a trusted, authenticated controller. This +// fences stale callers; it is not an authentication mechanism. Generations +// must increase across restores, rather than being generated by the guest. +const OwnerGenerationHeader = "ate-owner-generation" +const OwnerTokenHeader = "ate-owner-token" + +type ownerEpoch struct { + ctx context.Context + cancel context.CancelFunc + calls sync.WaitGroup +} + +type ownerFence struct { + mu sync.Mutex + generation int64 + token string + epoch *ownerEpoch + drain func() error +} + +func (f *ownerFence) acquire(ctx context.Context) (context.Context, func(), error) { + md, _ := metadata.FromIncomingContext(ctx) + generations, tokens := md.Get(OwnerGenerationHeader), md.Get(OwnerTokenHeader) + if len(generations) != 1 || len(tokens) != 1 || tokens[0] == "" || len(tokens[0]) > 256 { + return nil, nil, status.Error(codes.FailedPrecondition, "guest ownership metadata required") + } + generation, err := strconv.ParseInt(generations[0], 10, 64) + if err != nil || generation < 1 { + return nil, nil, status.Error(codes.InvalidArgument, "invalid owner generation") + } + f.mu.Lock() + defer f.mu.Unlock() + if generation < f.generation || generation == f.generation && tokens[0] != f.token { + return nil, nil, status.Error(codes.FailedPrecondition, "stale guest owner") + } + if generation == f.generation && f.epoch != nil && f.epoch.ctx.Err() != nil { + return nil, nil, status.Error(codes.FailedPrecondition, "guest owner is closed") + } + if generation > f.generation { + if f.epoch != nil { + f.epoch.cancel() + f.epoch.calls.Wait() + } + // No calls from the preceding epoch can spawn while the drain runs. + if err := f.drain(); err != nil { + return nil, nil, err + } + epochCtx, cancel := context.WithCancel(context.Background()) + f.epoch = &ownerEpoch{ctx: epochCtx, cancel: cancel} + f.generation, f.token = generation, tokens[0] + } + epoch := f.epoch + epoch.calls.Add(1) + callCtx, cancel := context.WithCancel(ctx) + stop := context.AfterFunc(epoch.ctx, cancel) + return callCtx, func() { stop(); cancel(); epoch.calls.Done() }, nil +} + +func (f *ownerFence) unary(ctx context.Context, req any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) { + ctx, release, err := f.acquire(ctx) + if err != nil { + return nil, err + } + defer release() + return handler(ctx, req) +} + +func (f *ownerFence) stream(srv any, stream grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error { + ctx, release, err := f.acquire(stream.Context()) + if err != nil { + return err + } + defer release() + return handler(srv, &ownedStream{ServerStream: stream, ctx: ctx}) +} + +type ownedStream struct { + grpc.ServerStream + ctx context.Context +} + +func (s *ownedStream) Context() context.Context { return s.ctx } + +// Returning from the handler closes the underlying transport, releasing an +// in-flight Send/Recv even when the old caller stopped consuming the stream. +func (s *ownedStream) operation(call func() error) error { + if err := s.ctx.Err(); err != nil { + return status.FromContextError(err).Err() + } + result := make(chan error, 1) + go func() { result <- call() }() + select { + case err := <-result: + return err + case <-s.ctx.Done(): + return status.FromContextError(s.ctx.Err()).Err() + } +} +func (s *ownedStream) SendMsg(m any) error { + return s.operation(func() error { return s.ServerStream.SendMsg(m) }) +} +func (s *ownedStream) RecvMsg(m any) error { + return s.operation(func() error { return s.ServerStream.RecvMsg(m) }) +} diff --git a/guest/ownership_test.go b/guest/ownership_test.go new file mode 100644 index 0000000..54a53cb --- /dev/null +++ b/guest/ownership_test.go @@ -0,0 +1,71 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package guest + +import ( + "context" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + "strconv" + "testing" + "time" +) + +func ownerContext(generation int, token string) context.Context { + return metadata.NewIncomingContext(context.Background(), metadata.Pairs(OwnerGenerationHeader, strconv.Itoa(generation), OwnerTokenHeader, token)) +} +func TestOwnerHandoffCancelsCallsBeforeDrainAndRejectsStaleOwners(t *testing.T) { + drains := 0 + fence := &ownerFence{drain: func() error { drains++; return nil }} + old, release, err := fence.acquire(ownerContext(1, "old")) + if err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { + _, release, err := fence.acquire(ownerContext(2, "new")) + if release != nil { + release() + } + done <- err + }() + select { + case <-old.Done(): + case <-time.After(time.Second): + t.Fatal("old call not canceled") + } + if drains != 1 { + t.Fatal("drained while old call still active") + } + release() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("handoff did not finish") + } + if drains != 2 { + t.Fatal("handoff did not drain") + } + for _, ctx := range []context.Context{ownerContext(1, "old"), ownerContext(2, "wrong"), context.Background()} { + _, _, err := fence.acquire(ctx) + if status.Code(err) != codes.FailedPrecondition { + t.Fatalf("stale owner admitted: %v", err) + } + } +} diff --git a/guest/process/service.go b/guest/process/service.go index a8d90e0..269b84f 100644 --- a/guest/process/service.go +++ b/guest/process/service.go @@ -16,13 +16,12 @@ package process import ( "context" - "errors" - "os" "time" ateenvv1alpha "github.com/agent-substrate/env/proto/ateenv/v1alpha" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/emptypb" ) // Service implements ateenvv1alpha.ProcessServiceServer. @@ -45,11 +44,20 @@ func (s *Service) StartProcess(ctx context.Context, req *ateenvv1alpha.StartProc return nil, status.Error(codes.InvalidArgument, "command cannot be empty") } - state, err := s.tracker.Start(req.GetCommand(), req.GetCwd(), req.GetEnv()) + if err := ctx.Err(); err != nil { + return nil, status.FromContextError(err).Err() + } + state, err := s.tracker.Start(req.GetCommand(), req.GetCwd(), req.GetEnv(), req.GetProcessId()) if err != nil { return nil, err } + if err := ctx.Err(); err != nil { + if _, killErr := s.tracker.Kill(state.ProcessID); killErr != nil { + return nil, killErr + } + return nil, status.FromContextError(err).Err() + } return &ateenvv1alpha.StartProcessResponse{ ProcessId: state.ProcessID, }, nil @@ -71,6 +79,9 @@ func (s *Service) GetProcess(ctx context.Context, req *ateenvv1alpha.GetProcessR // StreamProcessOutputs streams stdout and stderr in real-time or as a snapshot. func (s *Service) StreamProcessOutputs(req *ateenvv1alpha.StreamProcessOutputsRequest, stream ateenvv1alpha.ProcessService_StreamProcessOutputsServer) error { + if req.GetStdoutOffset() < 0 || req.GetStderrOffset() < 0 { + return status.Error(codes.InvalidArgument, "output offset cannot be negative") + } if req.GetProcessId() == "" { return status.Error(codes.InvalidArgument, "process_id cannot be empty") } @@ -116,33 +127,21 @@ func (s *Service) StreamProcessOutputs(req *ateenvv1alpha.StreamProcessOutputsRe stderrOffset = newStderrOffset } - if !follow { + if !follow && len(stdoutBytes) == 0 && len(stderrBytes) == 0 { // Snapshot mode: finish after reading available output up to this point return nil } // Check if process has finished and we consumed all output state.mu.RLock() - isTerminated := state.Status != ateenvv1alpha.ProcessStatus_PROCESS_STATUS_RUNNING + isTerminated := !state.FinishedAt.IsZero() state.mu.RUnlock() if isTerminated { - // Final check to see if there were any remaining bytes flushed on exit - finalStdout, _, _ := ReadLogs(state.StdoutPath, stdoutOffset) - if len(finalStdout) > 0 { - _ = stream.Send(&ateenvv1alpha.OutputChunk{ - Source: ateenvv1alpha.OutputSource_OUTPUT_SOURCE_STDOUT, - Data: finalStdout, - }) + if len(stdoutBytes) == 0 && len(stderrBytes) == 0 { + return nil } - finalStderr, _, _ := ReadLogs(state.StderrPath, stderrOffset) - if len(finalStderr) > 0 { - _ = stream.Send(&ateenvv1alpha.OutputChunk{ - Source: ateenvv1alpha.OutputSource_OUTPUT_SOURCE_STDERR, - Data: finalStderr, - }) - } - return nil + continue } // Sleep or wait for context cancellation @@ -162,13 +161,20 @@ func (s *Service) KillProcess(ctx context.Context, req *ateenvv1alpha.KillProces exitCode, err := s.tracker.Kill(req.GetProcessId()) if err != nil { - if errors.Is(err, os.ErrNotExist) { - return nil, status.Errorf(codes.NotFound, "process %q not found", req.GetProcessId()) - } - return nil, status.Errorf(codes.Internal, "killing process: %v", err) + return nil, err } return &ateenvv1alpha.KillProcessResponse{ ExitCode: exitCode, }, nil } + +func (s *Service) KillAllProcesses(ctx context.Context, _ *emptypb.Empty) (*emptypb.Empty, error) { + if err := ctx.Err(); err != nil { + return nil, status.FromContextError(err).Err() + } + if err := s.tracker.KillAll(); err != nil { + return nil, err + } + return &emptypb.Empty{}, nil +} diff --git a/guest/process/service_test.go b/guest/process/service_test.go index 02974d2..d835448 100644 --- a/guest/process/service_test.go +++ b/guest/process/service_test.go @@ -16,9 +16,12 @@ package process import ( "context" + "github.com/google/uuid" "io" + "math" "net" "os" + "path/filepath" "strings" "testing" "time" @@ -512,3 +515,92 @@ func TestWatchdogTimeout(t *testing.T) { t.Fatalf("expected exit code 137, got %d", proc.ExitCode) } } + +func TestCallerProcessIDCanBeCanceledBeforeStart(t *testing.T) { + client, cleanup := setupTestServer(t) + defer cleanup() + ctx := context.Background() + id := "b50b05c2-5c84-4d9b-bdcf-4016c35fc9d6" + _, err := client.KillProcess(ctx, &ateenvv1alpha.KillProcessRequest{ProcessId: id}) + if status.Code(err) != codes.NotFound { + t.Fatalf("kill before start: %v", err) + } + _, err = client.StartProcess(ctx, &ateenvv1alpha.StartProcessRequest{ProcessId: id, Command: []string{"sh", "-c", "exit 0"}}) + if status.Code(err) != codes.Canceled { + t.Fatalf("canceled ID started: %v", err) + } +} + +func TestOutputBacklogIsChunkedAndComplete(t *testing.T) { + client, cleanup := setupTestServer(t) + defer cleanup() + ctx := context.Background() + started, err := client.StartProcess(ctx, &ateenvv1alpha.StartProcessRequest{Command: []string{"sh", "-c", "head -c 5242880 /dev/zero"}}) + if err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(10 * time.Second) + for { + state, err := client.GetProcess(ctx, &ateenvv1alpha.GetProcessRequest{ProcessId: started.ProcessId}) + if err != nil { + t.Fatal(err) + } + if state.FinishedAt != nil { + break + } + if time.Now().After(deadline) { + t.Fatal("command did not complete") + } + time.Sleep(10 * time.Millisecond) + } + stream, err := client.StreamProcessOutputs(ctx, &ateenvv1alpha.StreamProcessOutputsRequest{ProcessId: started.ProcessId, Follow: true}) + if err != nil { + t.Fatal(err) + } + total := 0 + for { + chunk, err := stream.Recv() + if err == io.EOF { + break + } + if err != nil { + t.Fatal(err) + } + if len(chunk.Data) > 64*1024 { + t.Fatal("unbounded backlog chunk") + } + total += len(chunk.Data) + } + if total != 5242880 { + t.Fatalf("lost backlog tail: %d", total) + } +} + +func TestNegativeOutputOffsetsAndOwnerCancellationReset(t *testing.T) { + path := filepath.Join(t.TempDir(), "log") + if err := os.WriteFile(path, []byte("output"), 0600); err != nil { + t.Fatal(err) + } + for _, offset := range []int64{-1, math.MinInt64} { + if _, _, err := ReadLogs(path, offset); status.Code(err) != codes.InvalidArgument { + t.Fatalf("offset %d: %v", offset, err) + } + } + tracker, err := NewTracker(DefaultConfig(t.TempDir())) + if err != nil { + t.Fatal(err) + } + defer tracker.Close() + for i := 0; i < 1024; i++ { + tracker.Kill(uuid.NewString()) + } + if _, err := tracker.Start([]string{"sh", "-c", "true"}, "", nil); status.Code(err) != codes.ResourceExhausted { + t.Fatal("cancellation ceiling did not fail closed", err) + } + if err := tracker.ResetOwner(); err != nil { + t.Fatal(err) + } + if _, err := tracker.Start([]string{"sh", "-c", "true"}, "", nil); err != nil { + t.Fatal("new owner inherited previous cancellation ceiling", err) + } +} diff --git a/guest/process/tracker.go b/guest/process/tracker.go index 7773385..334ba71 100644 --- a/guest/process/tracker.go +++ b/guest/process/tracker.go @@ -23,11 +23,13 @@ import ( "os" "os/exec" "path/filepath" + "runtime" "sync" "syscall" "time" ateenvv1alpha "github.com/agent-substrate/env/proto/ateenv/v1alpha" + "github.com/google/uuid" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/types/known/timestamppb" @@ -54,6 +56,8 @@ type TrackerConfig struct { LogDir string // Workspace is the working and confinement directory for process operations. Workspace string + // ConfineWorkingDirectory holds a rooted directory descriptor through Linux startup. + ConfineWorkingDirectory bool // MaxConcurrentProcesses limits simultaneous active running commands. 0 means unlimited. MaxConcurrentProcesses int // MaxLogBytes caps stdout and stderr logs per command. 0 means unlimited. @@ -117,6 +121,7 @@ type Tracker struct { processes map[string]*ProcessState activeProcesses int stopPruner chan struct{} + cancelled map[string]bool } // NewTracker creates a new Process Tracker with Layer 1 resource controls. @@ -131,6 +136,7 @@ func NewTracker(cfg TrackerConfig) (*Tracker, error) { t := &Tracker{ config: cfg, processes: make(map[string]*ProcessState), + cancelled: make(map[string]bool), stopPruner: make(chan struct{}), } @@ -198,38 +204,94 @@ func (cw *cappedWriter) Write(p []byte) (n int, err error) { } // Start launches a new background process, enforcing concurrency and resource limits. -func (t *Tracker) Start(command []string, cwd string, env map[string]string) (*ProcessState, error) { +func (t *Tracker) Start(command []string, cwd string, env map[string]string, requestedID ...string) (*ProcessState, error) { if len(command) == 0 { return nil, status.Error(codes.InvalidArgument, "command list cannot be empty") } + processID := t.generateUniqueID() + if len(requestedID) > 0 && requestedID[0] != "" { + if _, err := uuid.Parse(requestedID[0]); err != nil { + return nil, status.Error(codes.InvalidArgument, "process_id must be a UUID") + } + processID = requestedID[0] + } t.mu.Lock() + defer t.mu.Unlock() + // ponytail: at most 1024 cancellation tombstones per owner; a new owner + // resets them only after the previous RPCs and processes have drained. + if len(t.cancelled) >= 1024 { + return nil, status.Error(codes.ResourceExhausted, "cancellation registry full") + } + if t.cancelled[processID] { + return nil, status.Error(codes.Canceled, "process was canceled before start") + } + if _, exists := t.processes[processID]; exists { + return nil, status.Error(codes.AlreadyExists, "process ID already used") + } if t.config.MaxConcurrentProcesses > 0 && t.activeProcesses >= t.config.MaxConcurrentProcesses { - t.mu.Unlock() return nil, status.Errorf(codes.ResourceExhausted, "maximum concurrent processes limit (%d) reached; please wait for running processes to complete or kill them", t.config.MaxConcurrentProcesses) } t.activeProcesses++ - t.mu.Unlock() - processID := t.generateUniqueID() stdoutPath := filepath.Join(t.config.LogDir, fmt.Sprintf("%s.stdout", processID)) stderrPath := filepath.Join(t.config.LogDir, fmt.Sprintf("%s.stderr", processID)) stdoutFile, err := os.OpenFile(stdoutPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644) if err != nil { - t.decrementActive() + t.activeProcesses-- return nil, status.Errorf(codes.Internal, "creating stdout log: %v", err) } stderrFile, err := os.OpenFile(stderrPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644) if err != nil { stdoutFile.Close() - t.decrementActive() + t.activeProcesses-- return nil, status.Errorf(codes.Internal, "creating stderr log: %v", err) } + launched := false + defer func() { + if !launched { + stdoutFile.Close() + stderrFile.Close() + } + }() + var directory *os.File + if t.config.ConfineWorkingDirectory && t.config.Workspace != "" && t.config.Workspace != "/" { + root, err := os.OpenRoot(t.config.Workspace) + if err != nil { + t.activeProcesses-- + return nil, status.Error(codes.Internal, "opening process workspace") + } + defer root.Close() + if cwd == "" { + cwd = t.config.Workspace + } + if !filepath.IsAbs(cwd) { + cwd = filepath.Join(t.config.Workspace, cwd) + } + relative, err := filepath.Rel(t.config.Workspace, cwd) + if err == nil { + directory, err = root.Open(relative) + } + if err != nil { + t.activeProcesses-- + return nil, status.Error(codes.PermissionDenied, "working directory outside workspace or unavailable") + } + defer directory.Close() + if info, err := directory.Stat(); err != nil || !info.IsDir() { + t.activeProcesses-- + return nil, status.Error(codes.InvalidArgument, "working directory is not a directory") + } + if runtime.GOOS != "linux" { + t.activeProcesses-- + return nil, status.Error(codes.Unimplemented, "confined process startup requires Linux") + } + cwd = fmt.Sprintf("/proc/%d/fd/%d", os.Getpid(), directory.Fd()) + } cmd := exec.Command(command[0], command[1:]...) if cwd != "" { cmd.Dir = cwd @@ -254,10 +316,11 @@ func (t *Tracker) Start(command []string, cwd string, env map[string]string) (*P if err := cmd.Start(); err != nil { stdoutFile.Close() stderrFile.Close() - t.decrementActive() + t.activeProcesses-- return nil, status.Errorf(codes.Internal, "starting process: %v", err) } + launched = true state := &ProcessState{ ProcessID: processID, Command: command, @@ -276,9 +339,7 @@ func (t *Tracker) Start(command []string, cwd string, env map[string]string) (*P }) } - t.mu.Lock() t.processes[processID] = state - t.mu.Unlock() // Background reaper goroutine go func() { @@ -351,43 +412,72 @@ func (t *Tracker) Get(processID string) (*ProcessState, bool) { // Kill terminates a running process and its process tree. func (t *Tracker) Kill(processID string) (int32, error) { - t.mu.RLock() + t.mu.Lock() state, ok := t.processes[processID] - t.mu.RUnlock() - if !ok { + if _, err := uuid.Parse(processID); err == nil { + if len(t.cancelled) >= 1024 && !t.cancelled[processID] { + t.mu.Unlock() + return 0, status.Error(codes.ResourceExhausted, "cancellation registry full") + } + t.cancelled[processID] = true + } + t.mu.Unlock() return 0, status.Errorf(codes.NotFound, "process %q not found", processID) } - + t.mu.Unlock() state.mu.Lock() - if state.Status != ateenvv1alpha.ProcessStatus_PROCESS_STATUS_RUNNING { - exitCode := state.ExitCode + if !state.FinishedAt.IsZero() { + code := state.ExitCode state.mu.Unlock() - return exitCode, nil + return code, nil + } + if err := syscall.Kill(-state.Cmd.Process.Pid, syscall.SIGKILL); err != nil && err != syscall.ESRCH { + state.mu.Unlock() + return 0, status.Error(codes.Internal, "process group did not terminate") } - state.Status = ateenvv1alpha.ProcessStatus_PROCESS_STATUS_TERMINATED - state.ExitCode = 128 + int32(syscall.SIGKILL) // 137 + state.ExitCode = 137 if state.timer != nil { state.timer.Stop() } - pid := state.Cmd.Process.Pid state.mu.Unlock() - - // Send SIGKILL to the entire process group (-PID) - _ = syscall.Kill(-pid, syscall.SIGKILL) - - // Wait for reaper goroutine select { case <-state.doneChan: case <-time.After(2 * time.Second): + return 0, status.Error(codes.DeadlineExceeded, "process termination unconfirmed") } - state.mu.RLock() - exitCode := state.ExitCode - state.mu.RUnlock() + defer state.mu.RUnlock() + return state.ExitCode, nil +} + +// KillAll drains tracked commands before changing an environment's owner. +func (t *Tracker) KillAll() error { + t.mu.RLock() + ids := make([]string, 0, len(t.processes)) + for id := range t.processes { + ids = append(ids, id) + } + t.mu.RUnlock() + for _, id := range ids { + if _, err := t.Kill(id); err != nil { + return err + } + } + return nil +} - return exitCode, nil +// ResetOwner drains the previous epoch before discarding its cancellation IDs. +// The caller must first fence and finish every RPC from that epoch. +func (t *Tracker) ResetOwner() error { + if err := t.KillAll(); err != nil { + return err + } + t.mu.Lock() + t.cancelled = make(map[string]bool) + t.mu.Unlock() + return nil } // prunerLoop periodically removes expired process states and log files. @@ -446,6 +536,9 @@ func (p *ProcessState) ToProto() *ateenvv1alpha.Process { // ReadLogs reads log bytes from a log file at a specific byte offset. func ReadLogs(filePath string, offset int64) ([]byte, int64, error) { + if offset < 0 { + return nil, 0, status.Error(codes.InvalidArgument, "output offset cannot be negative") + } f, err := os.Open(filePath) if err != nil { if errors.Is(err, os.ErrNotExist) { @@ -465,10 +558,13 @@ func ReadLogs(filePath string, offset int64) ([]byte, int64, error) { } length := size - offset + if length > 64*1024 { + length = 64 * 1024 + } buf := make([]byte, length) n, err := f.ReadAt(buf, offset) if err != nil && !errors.Is(err, io.EOF) { return nil, 0, err } - return buf[:n], size, nil + return buf[:n], offset + int64(n), nil } diff --git a/guest/server.go b/guest/server.go index ebc4b04..5ea5e58 100644 --- a/guest/server.go +++ b/guest/server.go @@ -44,6 +44,8 @@ type Config struct { EnableProcess bool // EnableFileSystem indicates whether the FileSystemService gRPC service is enabled. EnableFileSystem bool + // RequireOwner rejects calls without controller-issued ownership metadata. + RequireOwner bool } // DefaultConfig returns the default configuration with all services enabled. @@ -74,11 +76,17 @@ func FormatEnabledServices(cfg Config) string { func NewServer(cfg Config) (*grpc.Server, func(), error) { var cleanups []func() - grpcServer := grpc.NewServer() + fence := &ownerFence{drain: func() error { return nil }} + var opts []grpc.ServerOption + if cfg.RequireOwner { + opts = append(opts, grpc.UnaryInterceptor(fence.unary), grpc.StreamInterceptor(fence.stream)) + } + grpcServer := grpc.NewServer(opts...) reflection.Register(grpcServer) if cfg.EnableProcess { trackerCfg := process.DefaultConfig(cfg.LogDir) + trackerCfg.ConfineWorkingDirectory = cfg.RequireOwner if cfg.Workspace != "" { trackerCfg.Workspace = cfg.Workspace } @@ -86,6 +94,7 @@ func NewServer(cfg Config) (*grpc.Server, func(), error) { if err != nil { return nil, nil, fmt.Errorf("initializing process tracker: %w", err) } + fence.drain = tracker.ResetOwner cleanups = append(cleanups, func() { tracker.Close() }) diff --git a/proto/ateenv/v1alpha/guest.pb.go b/proto/ateenv/v1alpha/guest.pb.go index 53a1cb7..390826f 100644 --- a/proto/ateenv/v1alpha/guest.pb.go +++ b/proto/ateenv/v1alpha/guest.pb.go @@ -19,8 +19,8 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 -// protoc v5.28.2 +// protoc-gen-go v1.36.12 +// protoc (unknown) // source: proto/ateenv/v1alpha/guest.proto package ateenvv1alpha @@ -28,6 +28,7 @@ package ateenvv1alpha import ( protoreflect "google.golang.org/protobuf/reflect/protoreflect" protoimpl "google.golang.org/protobuf/runtime/protoimpl" + emptypb "google.golang.org/protobuf/types/known/emptypb" timestamppb "google.golang.org/protobuf/types/known/timestamppb" reflect "reflect" sync "sync" @@ -243,7 +244,10 @@ type StartProcessRequest struct { // Working directory inside the container (defaults to container workdir). Cwd string `protobuf:"bytes,2,opt,name=cwd,proto3" json:"cwd,omitempty"` // Environment variables to set for the background process. - Env map[string]string `protobuf:"bytes,3,rep,name=env,proto3" json:"env,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + Env map[string]string `protobuf:"bytes,3,rep,name=env,proto3" json:"env,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + // Optional caller-generated UUID. Allows cancellation before the start reply + // arrives; a canceled ID cannot subsequently launch a process. + ProcessId string `protobuf:"bytes,4,opt,name=process_id,json=processId,proto3" json:"process_id,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -299,6 +303,13 @@ func (x *StartProcessRequest) GetEnv() map[string]string { return nil } +func (x *StartProcessRequest) GetProcessId() string { + if x != nil { + return x.ProcessId + } + return "" +} + // Response returned immediately after launching an asynchronous process. type StartProcessResponse struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -813,11 +824,71 @@ func (x *WriteFileResponse) GetBytesWritten() int64 { return 0 } +type RemovePathRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + Path string `protobuf:"bytes,1,opt,name=path,proto3" json:"path,omitempty"` + Recursive bool `protobuf:"varint,2,opt,name=recursive,proto3" json:"recursive,omitempty"` + Force bool `protobuf:"varint,3,opt,name=force,proto3" json:"force,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *RemovePathRequest) Reset() { + *x = RemovePathRequest{} + mi := &file_proto_ateenv_v1alpha_guest_proto_msgTypes[12] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *RemovePathRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*RemovePathRequest) ProtoMessage() {} + +func (x *RemovePathRequest) ProtoReflect() protoreflect.Message { + mi := &file_proto_ateenv_v1alpha_guest_proto_msgTypes[12] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use RemovePathRequest.ProtoReflect.Descriptor instead. +func (*RemovePathRequest) Descriptor() ([]byte, []int) { + return file_proto_ateenv_v1alpha_guest_proto_rawDescGZIP(), []int{12} +} + +func (x *RemovePathRequest) GetPath() string { + if x != nil { + return x.Path + } + return "" +} + +func (x *RemovePathRequest) GetRecursive() bool { + if x != nil { + return x.Recursive + } + return false +} + +func (x *RemovePathRequest) GetForce() bool { + if x != nil { + return x.Force + } + return false +} + var File_proto_ateenv_v1alpha_guest_proto protoreflect.FileDescriptor const file_proto_ateenv_v1alpha_guest_proto_rawDesc = "" + "\n" + - " proto/ateenv/v1alpha/guest.proto\x12\x0eateenv.v1alpha\x1a\x1fgoogle/protobuf/timestamp.proto\"\xf4\x01\n" + + " proto/ateenv/v1alpha/guest.proto\x12\x0eateenv.v1alpha\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x1bgoogle/protobuf/empty.proto\"\xf4\x01\n" + "\aProcess\x12\x1d\n" + "\n" + "process_id\x18\x01 \x01(\tR\tprocessId\x125\n" + @@ -826,11 +897,13 @@ const file_proto_ateenv_v1alpha_guest_proto_rawDesc = "" + "\n" + "started_at\x18\x04 \x01(\v2\x1a.google.protobuf.TimestampR\tstartedAt\x12;\n" + "\vfinished_at\x18\x05 \x01(\v2\x1a.google.protobuf.TimestampR\n" + - "finishedAt\"\xb9\x01\n" + + "finishedAt\"\xd8\x01\n" + "\x13StartProcessRequest\x12\x18\n" + "\acommand\x18\x01 \x03(\tR\acommand\x12\x10\n" + "\x03cwd\x18\x02 \x01(\tR\x03cwd\x12>\n" + - "\x03env\x18\x03 \x03(\v2,.ateenv.v1alpha.StartProcessRequest.EnvEntryR\x03env\x1a6\n" + + "\x03env\x18\x03 \x03(\v2,.ateenv.v1alpha.StartProcessRequest.EnvEntryR\x03env\x12\x1d\n" + + "\n" + + "process_id\x18\x04 \x01(\tR\tprocessId\x1a6\n" + "\bEnvEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"5\n" + @@ -863,7 +936,11 @@ const file_proto_ateenv_v1alpha_guest_proto_rawDesc = "" + "\x05chunk\x18\x02 \x01(\fR\x05chunk\x12\x12\n" + "\x04mode\x18\x03 \x01(\rR\x04mode\"8\n" + "\x11WriteFileResponse\x12#\n" + - "\rbytes_written\x18\x01 \x01(\x03R\fbytesWritten*\xa3\x01\n" + + "\rbytes_written\x18\x01 \x01(\x03R\fbytesWritten\"[\n" + + "\x11RemovePathRequest\x12\x12\n" + + "\x04path\x18\x01 \x01(\tR\x04path\x12\x1c\n" + + "\trecursive\x18\x02 \x01(\bR\trecursive\x12\x14\n" + + "\x05force\x18\x03 \x01(\bR\x05force*\xa3\x01\n" + "\rProcessStatus\x12\x1e\n" + "\x1aPROCESS_STATUS_UNSPECIFIED\x10\x00\x12\x1a\n" + "\x16PROCESS_STATUS_RUNNING\x10\x01\x12\x1c\n" + @@ -873,16 +950,19 @@ const file_proto_ateenv_v1alpha_guest_proto_rawDesc = "" + "\fOutputSource\x12\x1d\n" + "\x19OUTPUT_SOURCE_UNSPECIFIED\x10\x00\x12\x18\n" + "\x14OUTPUT_SOURCE_STDOUT\x10\x01\x12\x18\n" + - "\x14OUTPUT_SOURCE_STDERR\x10\x022\xf1\x02\n" + + "\x14OUTPUT_SOURCE_STDERR\x10\x022\xb5\x03\n" + "\x0eProcessService\x12Y\n" + "\fStartProcess\x12#.ateenv.v1alpha.StartProcessRequest\x1a$.ateenv.v1alpha.StartProcessResponse\x12H\n" + "\n" + "GetProcess\x12!.ateenv.v1alpha.GetProcessRequest\x1a\x17.ateenv.v1alpha.Process\x12b\n" + "\x14StreamProcessOutputs\x12+.ateenv.v1alpha.StreamProcessOutputsRequest\x1a\x1b.ateenv.v1alpha.OutputChunk0\x01\x12V\n" + - "\vKillProcess\x12\".ateenv.v1alpha.KillProcessRequest\x1a#.ateenv.v1alpha.KillProcessResponse2\xb1\x01\n" + + "\vKillProcess\x12\".ateenv.v1alpha.KillProcessRequest\x1a#.ateenv.v1alpha.KillProcessResponse\x12B\n" + + "\x10KillAllProcesses\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.Empty2\xfa\x01\n" + "\x11FileSystemService\x12H\n" + "\bReadFile\x12\x1f.ateenv.v1alpha.ReadFileRequest\x1a\x19.ateenv.v1alpha.FileChunk0\x01\x12R\n" + - "\tWriteFile\x12 .ateenv.v1alpha.WriteFileRequest\x1a!.ateenv.v1alpha.WriteFileResponse(\x01BCZAgithub.com/agent-substrate/env/proto/ateenv/v1alpha;ateenvv1alphab\x06proto3" + "\tWriteFile\x12 .ateenv.v1alpha.WriteFileRequest\x1a!.ateenv.v1alpha.WriteFileResponse(\x01\x12G\n" + + "\n" + + "RemovePath\x12!.ateenv.v1alpha.RemovePathRequest\x1a\x16.google.protobuf.EmptyBCZAgithub.com/agent-substrate/env/proto/ateenv/v1alpha;ateenvv1alphab\x06proto3" var ( file_proto_ateenv_v1alpha_guest_proto_rawDescOnce sync.Once @@ -897,7 +977,7 @@ func file_proto_ateenv_v1alpha_guest_proto_rawDescGZIP() []byte { } var file_proto_ateenv_v1alpha_guest_proto_enumTypes = make([]protoimpl.EnumInfo, 2) -var file_proto_ateenv_v1alpha_guest_proto_msgTypes = make([]protoimpl.MessageInfo, 13) +var file_proto_ateenv_v1alpha_guest_proto_msgTypes = make([]protoimpl.MessageInfo, 14) var file_proto_ateenv_v1alpha_guest_proto_goTypes = []any{ (ProcessStatus)(0), // 0: ateenv.v1alpha.ProcessStatus (OutputSource)(0), // 1: ateenv.v1alpha.OutputSource @@ -913,29 +993,35 @@ var file_proto_ateenv_v1alpha_guest_proto_goTypes = []any{ (*FileChunk)(nil), // 11: ateenv.v1alpha.FileChunk (*WriteFileRequest)(nil), // 12: ateenv.v1alpha.WriteFileRequest (*WriteFileResponse)(nil), // 13: ateenv.v1alpha.WriteFileResponse - nil, // 14: ateenv.v1alpha.StartProcessRequest.EnvEntry - (*timestamppb.Timestamp)(nil), // 15: google.protobuf.Timestamp + (*RemovePathRequest)(nil), // 14: ateenv.v1alpha.RemovePathRequest + nil, // 15: ateenv.v1alpha.StartProcessRequest.EnvEntry + (*timestamppb.Timestamp)(nil), // 16: google.protobuf.Timestamp + (*emptypb.Empty)(nil), // 17: google.protobuf.Empty } var file_proto_ateenv_v1alpha_guest_proto_depIdxs = []int32{ 0, // 0: ateenv.v1alpha.Process.status:type_name -> ateenv.v1alpha.ProcessStatus - 15, // 1: ateenv.v1alpha.Process.started_at:type_name -> google.protobuf.Timestamp - 15, // 2: ateenv.v1alpha.Process.finished_at:type_name -> google.protobuf.Timestamp - 14, // 3: ateenv.v1alpha.StartProcessRequest.env:type_name -> ateenv.v1alpha.StartProcessRequest.EnvEntry + 16, // 1: ateenv.v1alpha.Process.started_at:type_name -> google.protobuf.Timestamp + 16, // 2: ateenv.v1alpha.Process.finished_at:type_name -> google.protobuf.Timestamp + 15, // 3: ateenv.v1alpha.StartProcessRequest.env:type_name -> ateenv.v1alpha.StartProcessRequest.EnvEntry 1, // 4: ateenv.v1alpha.OutputChunk.source:type_name -> ateenv.v1alpha.OutputSource 3, // 5: ateenv.v1alpha.ProcessService.StartProcess:input_type -> ateenv.v1alpha.StartProcessRequest 5, // 6: ateenv.v1alpha.ProcessService.GetProcess:input_type -> ateenv.v1alpha.GetProcessRequest 6, // 7: ateenv.v1alpha.ProcessService.StreamProcessOutputs:input_type -> ateenv.v1alpha.StreamProcessOutputsRequest 8, // 8: ateenv.v1alpha.ProcessService.KillProcess:input_type -> ateenv.v1alpha.KillProcessRequest - 10, // 9: ateenv.v1alpha.FileSystemService.ReadFile:input_type -> ateenv.v1alpha.ReadFileRequest - 12, // 10: ateenv.v1alpha.FileSystemService.WriteFile:input_type -> ateenv.v1alpha.WriteFileRequest - 4, // 11: ateenv.v1alpha.ProcessService.StartProcess:output_type -> ateenv.v1alpha.StartProcessResponse - 2, // 12: ateenv.v1alpha.ProcessService.GetProcess:output_type -> ateenv.v1alpha.Process - 7, // 13: ateenv.v1alpha.ProcessService.StreamProcessOutputs:output_type -> ateenv.v1alpha.OutputChunk - 9, // 14: ateenv.v1alpha.ProcessService.KillProcess:output_type -> ateenv.v1alpha.KillProcessResponse - 11, // 15: ateenv.v1alpha.FileSystemService.ReadFile:output_type -> ateenv.v1alpha.FileChunk - 13, // 16: ateenv.v1alpha.FileSystemService.WriteFile:output_type -> ateenv.v1alpha.WriteFileResponse - 11, // [11:17] is the sub-list for method output_type - 5, // [5:11] is the sub-list for method input_type + 17, // 9: ateenv.v1alpha.ProcessService.KillAllProcesses:input_type -> google.protobuf.Empty + 10, // 10: ateenv.v1alpha.FileSystemService.ReadFile:input_type -> ateenv.v1alpha.ReadFileRequest + 12, // 11: ateenv.v1alpha.FileSystemService.WriteFile:input_type -> ateenv.v1alpha.WriteFileRequest + 14, // 12: ateenv.v1alpha.FileSystemService.RemovePath:input_type -> ateenv.v1alpha.RemovePathRequest + 4, // 13: ateenv.v1alpha.ProcessService.StartProcess:output_type -> ateenv.v1alpha.StartProcessResponse + 2, // 14: ateenv.v1alpha.ProcessService.GetProcess:output_type -> ateenv.v1alpha.Process + 7, // 15: ateenv.v1alpha.ProcessService.StreamProcessOutputs:output_type -> ateenv.v1alpha.OutputChunk + 9, // 16: ateenv.v1alpha.ProcessService.KillProcess:output_type -> ateenv.v1alpha.KillProcessResponse + 17, // 17: ateenv.v1alpha.ProcessService.KillAllProcesses:output_type -> google.protobuf.Empty + 11, // 18: ateenv.v1alpha.FileSystemService.ReadFile:output_type -> ateenv.v1alpha.FileChunk + 13, // 19: ateenv.v1alpha.FileSystemService.WriteFile:output_type -> ateenv.v1alpha.WriteFileResponse + 17, // 20: ateenv.v1alpha.FileSystemService.RemovePath:output_type -> google.protobuf.Empty + 13, // [13:21] is the sub-list for method output_type + 5, // [5:13] is the sub-list for method input_type 5, // [5:5] is the sub-list for extension type_name 5, // [5:5] is the sub-list for extension extendee 0, // [0:5] is the sub-list for field type_name @@ -952,7 +1038,7 @@ func file_proto_ateenv_v1alpha_guest_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_ateenv_v1alpha_guest_proto_rawDesc), len(file_proto_ateenv_v1alpha_guest_proto_rawDesc)), NumEnums: 2, - NumMessages: 13, + NumMessages: 14, NumExtensions: 0, NumServices: 2, }, diff --git a/proto/ateenv/v1alpha/guest.proto b/proto/ateenv/v1alpha/guest.proto index b979b7a..63c8b67 100644 --- a/proto/ateenv/v1alpha/guest.proto +++ b/proto/ateenv/v1alpha/guest.proto @@ -24,6 +24,7 @@ package ateenv.v1alpha; option go_package = "github.com/agent-substrate/env/proto/ateenv/v1alpha;ateenvv1alpha"; import "google/protobuf/timestamp.proto"; +import "google/protobuf/empty.proto"; // ============================================================================ // --- SERVICES --- @@ -44,6 +45,9 @@ service ProcessService { // KillProcess terminates a running asynchronous process and its child process tree. rpc KillProcess(KillProcessRequest) returns (KillProcessResponse); + + // Drain tracked processes before suspending an environment. + rpc KillAllProcesses(google.protobuf.Empty) returns (google.protobuf.Empty); } // FileSystemService provides streaming file reading and writing capabilities inside @@ -54,6 +58,9 @@ service FileSystemService { // WriteFile streams binary or text chunks directly to a target file. rpc WriteFile(stream WriteFileRequest) returns (WriteFileResponse); + + // Remove a file or directory within the configured filesystem root. + rpc RemovePath(RemovePathRequest) returns (google.protobuf.Empty); } // ============================================================================ @@ -104,6 +111,9 @@ message StartProcessRequest { string cwd = 2; // Environment variables to set for the background process. map env = 3; + // Optional caller-generated UUID. Allows cancellation before the start reply + // arrives; a canceled ID cannot subsequently launch a process. + string process_id = 4; } // Response returned immediately after launching an asynchronous process. @@ -181,3 +191,9 @@ message WriteFileResponse { // Total number of bytes written across all stream chunks. int64 bytes_written = 1; } + +message RemovePathRequest { + string path = 1; + bool recursive = 2; + bool force = 3; +} diff --git a/proto/ateenv/v1alpha/guest_grpc.pb.go b/proto/ateenv/v1alpha/guest_grpc.pb.go index 88af382..f38e55b 100644 --- a/proto/ateenv/v1alpha/guest_grpc.pb.go +++ b/proto/ateenv/v1alpha/guest_grpc.pb.go @@ -20,7 +20,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.5.1 -// - protoc v5.28.2 +// - protoc (unknown) // source: proto/ateenv/v1alpha/guest.proto package ateenvv1alpha @@ -30,6 +30,7 @@ import ( grpc "google.golang.org/grpc" codes "google.golang.org/grpc/codes" status "google.golang.org/grpc/status" + emptypb "google.golang.org/protobuf/types/known/emptypb" ) // This is a compile-time assertion to ensure that this generated file @@ -42,6 +43,7 @@ const ( ProcessService_GetProcess_FullMethodName = "/ateenv.v1alpha.ProcessService/GetProcess" ProcessService_StreamProcessOutputs_FullMethodName = "/ateenv.v1alpha.ProcessService/StreamProcessOutputs" ProcessService_KillProcess_FullMethodName = "/ateenv.v1alpha.ProcessService/KillProcess" + ProcessService_KillAllProcesses_FullMethodName = "/ateenv.v1alpha.ProcessService/KillAllProcesses" ) // ProcessServiceClient is the client API for ProcessService service. @@ -60,6 +62,8 @@ type ProcessServiceClient interface { StreamProcessOutputs(ctx context.Context, in *StreamProcessOutputsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[OutputChunk], error) // KillProcess terminates a running asynchronous process and its child process tree. KillProcess(ctx context.Context, in *KillProcessRequest, opts ...grpc.CallOption) (*KillProcessResponse, error) + // Drain tracked processes before suspending an environment. + KillAllProcesses(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*emptypb.Empty, error) } type processServiceClient struct { @@ -119,6 +123,16 @@ func (c *processServiceClient) KillProcess(ctx context.Context, in *KillProcessR return out, nil } +func (c *processServiceClient) KillAllProcesses(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*emptypb.Empty, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(emptypb.Empty) + err := c.cc.Invoke(ctx, ProcessService_KillAllProcesses_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + // ProcessServiceServer is the server API for ProcessService service. // All implementations must embed UnimplementedProcessServiceServer // for forward compatibility. @@ -135,6 +149,8 @@ type ProcessServiceServer interface { StreamProcessOutputs(*StreamProcessOutputsRequest, grpc.ServerStreamingServer[OutputChunk]) error // KillProcess terminates a running asynchronous process and its child process tree. KillProcess(context.Context, *KillProcessRequest) (*KillProcessResponse, error) + // Drain tracked processes before suspending an environment. + KillAllProcesses(context.Context, *emptypb.Empty) (*emptypb.Empty, error) mustEmbedUnimplementedProcessServiceServer() } @@ -157,6 +173,9 @@ func (UnimplementedProcessServiceServer) StreamProcessOutputs(*StreamProcessOutp func (UnimplementedProcessServiceServer) KillProcess(context.Context, *KillProcessRequest) (*KillProcessResponse, error) { return nil, status.Errorf(codes.Unimplemented, "method KillProcess not implemented") } +func (UnimplementedProcessServiceServer) KillAllProcesses(context.Context, *emptypb.Empty) (*emptypb.Empty, error) { + return nil, status.Errorf(codes.Unimplemented, "method KillAllProcesses not implemented") +} func (UnimplementedProcessServiceServer) mustEmbedUnimplementedProcessServiceServer() {} func (UnimplementedProcessServiceServer) testEmbeddedByValue() {} @@ -243,6 +262,24 @@ func _ProcessService_KillProcess_Handler(srv interface{}, ctx context.Context, d return interceptor(ctx, in, info, handler) } +func _ProcessService_KillAllProcesses_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(emptypb.Empty) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ProcessServiceServer).KillAllProcesses(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: ProcessService_KillAllProcesses_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ProcessServiceServer).KillAllProcesses(ctx, req.(*emptypb.Empty)) + } + return interceptor(ctx, in, info, handler) +} + // ProcessService_ServiceDesc is the grpc.ServiceDesc for ProcessService service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -262,6 +299,10 @@ var ProcessService_ServiceDesc = grpc.ServiceDesc{ MethodName: "KillProcess", Handler: _ProcessService_KillProcess_Handler, }, + { + MethodName: "KillAllProcesses", + Handler: _ProcessService_KillAllProcesses_Handler, + }, }, Streams: []grpc.StreamDesc{ { @@ -274,8 +315,9 @@ var ProcessService_ServiceDesc = grpc.ServiceDesc{ } const ( - FileSystemService_ReadFile_FullMethodName = "/ateenv.v1alpha.FileSystemService/ReadFile" - FileSystemService_WriteFile_FullMethodName = "/ateenv.v1alpha.FileSystemService/WriteFile" + FileSystemService_ReadFile_FullMethodName = "/ateenv.v1alpha.FileSystemService/ReadFile" + FileSystemService_WriteFile_FullMethodName = "/ateenv.v1alpha.FileSystemService/WriteFile" + FileSystemService_RemovePath_FullMethodName = "/ateenv.v1alpha.FileSystemService/RemovePath" ) // FileSystemServiceClient is the client API for FileSystemService service. @@ -289,6 +331,8 @@ type FileSystemServiceClient interface { ReadFile(ctx context.Context, in *ReadFileRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[FileChunk], error) // WriteFile streams binary or text chunks directly to a target file. WriteFile(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[WriteFileRequest, WriteFileResponse], error) + // Remove a file or directory within the configured filesystem root. + RemovePath(ctx context.Context, in *RemovePathRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) } type fileSystemServiceClient struct { @@ -331,6 +375,16 @@ func (c *fileSystemServiceClient) WriteFile(ctx context.Context, opts ...grpc.Ca // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type FileSystemService_WriteFileClient = grpc.ClientStreamingClient[WriteFileRequest, WriteFileResponse] +func (c *fileSystemServiceClient) RemovePath(ctx context.Context, in *RemovePathRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(emptypb.Empty) + err := c.cc.Invoke(ctx, FileSystemService_RemovePath_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + // FileSystemServiceServer is the server API for FileSystemService service. // All implementations must embed UnimplementedFileSystemServiceServer // for forward compatibility. @@ -342,6 +396,8 @@ type FileSystemServiceServer interface { ReadFile(*ReadFileRequest, grpc.ServerStreamingServer[FileChunk]) error // WriteFile streams binary or text chunks directly to a target file. WriteFile(grpc.ClientStreamingServer[WriteFileRequest, WriteFileResponse]) error + // Remove a file or directory within the configured filesystem root. + RemovePath(context.Context, *RemovePathRequest) (*emptypb.Empty, error) mustEmbedUnimplementedFileSystemServiceServer() } @@ -358,6 +414,9 @@ func (UnimplementedFileSystemServiceServer) ReadFile(*ReadFileRequest, grpc.Serv func (UnimplementedFileSystemServiceServer) WriteFile(grpc.ClientStreamingServer[WriteFileRequest, WriteFileResponse]) error { return status.Errorf(codes.Unimplemented, "method WriteFile not implemented") } +func (UnimplementedFileSystemServiceServer) RemovePath(context.Context, *RemovePathRequest) (*emptypb.Empty, error) { + return nil, status.Errorf(codes.Unimplemented, "method RemovePath not implemented") +} func (UnimplementedFileSystemServiceServer) mustEmbedUnimplementedFileSystemServiceServer() {} func (UnimplementedFileSystemServiceServer) testEmbeddedByValue() {} @@ -397,13 +456,36 @@ func _FileSystemService_WriteFile_Handler(srv interface{}, stream grpc.ServerStr // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type FileSystemService_WriteFileServer = grpc.ClientStreamingServer[WriteFileRequest, WriteFileResponse] +func _FileSystemService_RemovePath_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(RemovePathRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(FileSystemServiceServer).RemovePath(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: FileSystemService_RemovePath_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(FileSystemServiceServer).RemovePath(ctx, req.(*RemovePathRequest)) + } + return interceptor(ctx, in, info, handler) +} + // FileSystemService_ServiceDesc is the grpc.ServiceDesc for FileSystemService service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) var FileSystemService_ServiceDesc = grpc.ServiceDesc{ ServiceName: "ateenv.v1alpha.FileSystemService", HandlerType: (*FileSystemServiceServer)(nil), - Methods: []grpc.MethodDesc{}, + Methods: []grpc.MethodDesc{ + { + MethodName: "RemovePath", + Handler: _FileSystemService_RemovePath_Handler, + }, + }, Streams: []grpc.StreamDesc{ { StreamName: "ReadFile",