soma_infra/
process_build_context.rs

1use std::path::Path;
2use std::sync::Arc;
3
4use async_trait::async_trait;
5use base64::Engine;
6use serde::Deserialize;
7use soma_fleet::{CommandExecutor, CommandOutput, CommandRequest, HostRecord};
8use tokio_util::sync::CancellationToken;
9
10use crate::{
11    BuildContextFingerprint, BuildContextInspector, BuildContextPolicy, InfraError, InfraResult,
12};
13
14const OUTPUT_LIMIT: usize = 64 * 1024;
15const SCRIPT: &str = r#"import hashlib,json,os,stat,sys
16root,rel,max_files,max_bytes=sys.argv[1],sys.argv[2],int(sys.argv[3]),int(sys.argv[4])
17def open_beneath(root,rel):
18 fd=os.open('/',os.O_RDONLY|os.O_DIRECTORY)
19 try:
20  for part in [p for p in root.split('/') if p]+[p for p in rel.split('/') if p and p!='.']:
21   nxt=os.open(part,os.O_RDONLY|os.O_NOFOLLOW,dir_fd=fd)
22   os.close(fd); fd=nxt
23  return fd
24 except Exception:
25  os.close(fd); raise
26h=hashlib.sha256(); counts=[0,0]
27def feed(kind,path,meta):
28 data=(kind+'\0'+path+'\0'+oct(meta.st_mode & 0o777)+'\0'+str(meta.st_size)+'\0').encode()
29 h.update(len(data).to_bytes(8,'big')); h.update(data)
30def walk(fd,path):
31 meta=os.fstat(fd)
32 if stat.S_ISDIR(meta.st_mode):
33  feed('d',path,meta)
34  with os.scandir(fd) as entries:
35   for entry in sorted(entries,key=lambda e:e.name):
36    child=os.open(entry.name,os.O_RDONLY|os.O_NOFOLLOW,dir_fd=fd)
37    try: walk(child,entry.name if path=='.' else path+'/'+entry.name)
38    finally: os.close(child)
39 elif stat.S_ISREG(meta.st_mode):
40  counts[0]+=1; counts[1]+=meta.st_size
41  if counts[0]>max_files: raise RuntimeError('build context file limit exceeded')
42  if counts[1]>max_bytes: raise RuntimeError('build context byte limit exceeded')
43  feed('f',path,meta); os.lseek(fd,0,os.SEEK_SET)
44  while True:
45   chunk=os.read(fd,1024*1024)
46   if not chunk: break
47   h.update(chunk)
48 else: raise RuntimeError('build context contains symlink or unsupported file type')
49fd=open_beneath(root,rel)
50try:
51 if not stat.S_ISDIR(os.fstat(fd).st_mode): raise RuntimeError('build context is not a directory')
52 walk(fd,'.')
53 print(json.dumps({'sha256':h.hexdigest(),'file_count':counts[0],'byte_count':counts[1]}))
54finally: os.close(fd)
55"#;
56
57/// Descriptor-confined build-context inspector backed by fleet command execution.
58pub struct CommandBuildContextInspector<E> {
59    executor: Arc<E>,
60    policy: BuildContextPolicy,
61}
62
63impl<E> CommandBuildContextInspector<E> {
64    /// Creates an inspector from an executor and explicit policy.
65    #[must_use]
66    pub fn new(executor: Arc<E>, policy: BuildContextPolicy) -> Self {
67        Self { executor, policy }
68    }
69
70    /// Returns the active build-context policy.
71    #[must_use]
72    pub const fn policy(&self) -> &BuildContextPolicy {
73        &self.policy
74    }
75}
76
77#[async_trait]
78impl<E> BuildContextInspector for CommandBuildContextInspector<E>
79where
80    E: CommandExecutor,
81{
82    async fn fingerprint(
83        &self,
84        host: &HostRecord,
85        path: &Path,
86        deadline: soma_ops::Timestamp,
87        cancellation: &CancellationToken,
88    ) -> InfraResult<BuildContextFingerprint> {
89        let (root, relative) = self.policy.resolve(path)?;
90        let encoded = base64::engine::general_purpose::STANDARD.encode(SCRIPT.as_bytes());
91        let bootstrap = format!("import base64;exec(base64.b64decode('{encoded}'))");
92        let relative = if relative.as_os_str().is_empty() {
93            ".".into()
94        } else {
95            relative.to_string_lossy().into_owned()
96        };
97        let request = CommandRequest::new(
98            "python3",
99            [
100                "-c".to_owned(),
101                bootstrap,
102                root.to_string_lossy().into_owned(),
103                relative,
104                self.policy.max_files().to_string(),
105                self.policy.max_bytes().to_string(),
106            ],
107            deadline,
108        )
109        .map_err(soma_fleet::FleetError::from)?
110        .with_output_limits(OUTPUT_LIMIT, OUTPUT_LIMIT)
111        .map_err(soma_fleet::FleetError::from)?;
112        let wire = parse_output(
113            host,
114            self.executor.execute(host, &request, cancellation).await?,
115        )?;
116        let result = BuildContextFingerprint {
117            host: host.id().clone(),
118            topology_revision: host.revision().clone(),
119            path: path.to_path_buf(),
120            sha256: wire.sha256,
121            file_count: wire.file_count,
122            byte_count: wire.byte_count,
123        };
124        result.validate()?;
125        Ok(result)
126    }
127}
128
129#[derive(Deserialize)]
130struct FingerprintWire {
131    sha256: String,
132    file_count: u32,
133    byte_count: u64,
134}
135
136fn parse_output(host: &HostRecord, output: CommandOutput) -> InfraResult<FingerprintWire> {
137    if output.exit_code() != Some(0) {
138        return Err(InfraError::CommandFailed {
139            domain: "build-context",
140            host: host.id().clone(),
141            exit_code: output.exit_code(),
142            stderr: String::from_utf8_lossy(output.stderr()).trim().to_owned(),
143        });
144    }
145    if output.truncated() {
146        return Err(InfraError::Parse {
147            domain: "build-context",
148            message: "fingerprint output was truncated".into(),
149        });
150    }
151    serde_json::from_slice(output.stdout()).map_err(|error| InfraError::Parse {
152        domain: "build-context",
153        message: format!("invalid fingerprint payload: {error}"),
154    })
155}
156
157#[cfg(test)]
158#[path = "process_build_context_tests.rs"]
159mod tests;
soma_infra/process_build_context.rs

soma_infra/
process_build_context.rs
