Menu
akurai-tasks
publicLatest change 18f4b1ebc4782165ffc2d599bc25ac07a245be76 - Harden task mutations and repair web flows by Ólafur Búi Ólafsson
use super::*;
use sha2::{Digest, Sha256};
use std::fs;
pub fn discover_sources(root: &Path) -> Result<Value> {
let mut boards = Vec::new();
discover(root, &mut boards)?;
boards.sort();
Ok(object(vec![
("version", Value::Int(1)),
("root", string(&root.display().to_string())),
(
"sources",
Value::Array(
boards
.iter()
.map(|path| {
object(vec![
("kind", string("file-kanban")),
("location", string(path)),
(
"disposition",
string(if path.contains("github-workflow-audit") {
"snapshot"
} else {
"live"
}),
),
])
})
.collect(),
),
),
("unresolved", Value::Array(Vec::new())),
("createdAt", Value::Int(now())),
]))
}
pub fn plan_file_board(board: &Path, project: &str, role: &str) -> Result<Value> {
validate_key(project)?;
if !matches!(role, "live" | "snapshot") {
return Err(Error::Invalid(
"source role must be live or snapshot".into(),
));
}
let canonical = fs::canonicalize(board).map_err(storage)?;
let mut files = Vec::new();
collect_cards(&canonical, &mut files)?;
files.sort();
let mut entities = Vec::new();
let mut quarantined = 0i64;
for path in files {
let bytes = fs::read(&path).map_err(storage)?;
let source_key = path
.strip_prefix(&canonical)
.unwrap_or(&path)
.display()
.to_string();
let digest = bytes_digest(&bytes);
let secret = contains_secret(&bytes);
if secret {
quarantined += 1;
}
let text = String::from_utf8_lossy(&bytes);
let title = text
.lines()
.find_map(|line| line.trim().strip_prefix("# "))
.unwrap_or_else(|| {
path.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("Imported item")
});
let state = state_from_path(&source_key);
entities.push(object(vec![
("sourceKey", string(&source_key)),
("sourcePath", string(&path.display().to_string())),
("title", string(title.trim())),
("state", string(&state)),
("bytes", Value::Int(bytes.len() as i64)),
("sha256", string(&digest)),
(
"rawBase64",
if secret {
Value::Null
} else {
string(&base64(&bytes))
},
),
(
"outcome",
string(if secret {
"quarantined"
} else if role == "snapshot" {
"archived"
} else {
"imported"
}),
),
(
"diagnostic",
if secret {
string("credential-shaped content; raw bytes not copied")
} else {
Value::Null
},
),
]));
}
let source = canonical.display().to_string();
let bundle_without_digest = object(vec![
("version", Value::Int(1)),
("kind", string("file-kanban")),
("source", string(&source)),
("role", string(role)),
("project", string(project)),
("entities", Value::Array(entities)),
("quarantined", Value::Int(quarantined)),
("planDigest", Value::Null),
]);
let digest = auth::digest(&bundle_without_digest.to_json());
let mut bundle = bundle_without_digest;
update(&mut bundle, "planDigest", string(&digest));
Ok(bundle)
}
impl Store {
pub fn apply_import(&self, bundle: &Value, actor: &str, idempotency: &str) -> Result<Value> {
if bundle.get("version").and_then(Value::as_i64) != Some(1) {
return Err(Error::Invalid("unsupported import bundle version".into()));
}
let source = field(bundle, "source")
.ok_or_else(|| Error::Invalid("bundle source missing".into()))?;
let project = field(bundle, "project")
.ok_or_else(|| Error::Invalid("bundle project missing".into()))?;
let role =
field(bundle, "role").ok_or_else(|| Error::Invalid("bundle role missing".into()))?;
let digest = field(bundle, "planDigest")
.ok_or_else(|| Error::Invalid("bundle plan digest missing".into()))?;
let mut canonical = bundle.clone();
update(&mut canonical, "planDigest", Value::Null);
if auth::digest(&canonical.to_json()) != digest {
return Err(Error::Conflict("import plan digest mismatch".into()));
}
let entities = match bundle.get("entities") {
Some(Value::Array(values)) => values.clone(),
_ => return Err(Error::Invalid("bundle entities missing".into())),
};
let request = object(vec![
("command", string("import_apply")),
("source", string(source)),
("project", string(project)),
("planDigest", string(digest)),
]);
self.mutate(actor, idempotency, &request, |db, sequence| {
let run_id = digest;
if let Some(existing) = get_value(db, &format!("import-run/{run_id}"))? {
return Ok(existing);
}
if get_value(db, &format!("project/{project}"))?.is_none() {
let project_record = object(vec![
("id", string(project)),
("key", string(project)),
("name", string(project)),
("repos", Value::Array(Vec::new())),
("archived", Value::Bool(false)),
("revision", Value::Int(1)),
("createdAt", Value::Int(now())),
("updatedAt", Value::Int(now())),
]);
put(db, &format!("project/{project}"), &project_record)?;
put(
db,
&format!("project-order/{sequence:020}/{project}"),
&project_record,
)?;
}
let source_id = auth::digest(source);
let mut imported = 0i64;
let mut archived = 0i64;
let mut quarantined = 0i64;
for entity in &entities {
let source_key = field(entity, "sourceKey")
.ok_or_else(|| Error::Invalid("source key missing".into()))?;
let outcome = field(entity, "outcome").unwrap_or("quarantined");
let entity_id = auth::digest(&format!("{source}\0{source_key}"));
let raw_ref = format!("import-raw/{run_id}/{entity_id}");
if let Some(raw) = field(entity, "rawBase64") {
let raw_record = object(vec![
("source", string(source)),
("sourceKey", string(source_key)),
("base64", string(raw)),
(
"bytes",
entity.get("bytes").cloned().unwrap_or(Value::Int(0)),
),
(
"sha256",
entity.get("sha256").cloned().unwrap_or(Value::Null),
),
]);
put(db, &raw_ref, &raw_record)?;
}
let destination = match outcome {
"imported" => {
imported += 1;
let item_sequence = next_counter(db, &format!("counter/item/{project}"))?;
let id = format!("{project}-{item_sequence}");
let state = field(entity, "state")
.filter(|state| STATES.contains(state))
.unwrap_or("Inbox");
let item = object(vec![
("id", string(&id)),
("project", string(project)),
("sequence", Value::Int(item_sequence)),
(
"title",
string(field(entity, "title").unwrap_or(source_key)),
),
("description", string("")),
("repo", Value::Null),
("state", string(state)),
("priority", string("normal")),
("owner", Value::Null),
("leaseGeneration", Value::Int(0)),
("blocked", Value::Bool(false)),
("blockReason", Value::Null),
("sourceInstance", string(&source_id)),
("sourceKey", string(source_key)),
(
"rawRef",
if field(entity, "rawBase64").is_some() {
string(&raw_ref)
} else {
Value::Null
},
),
("revision", Value::Int(1)),
("createdAt", Value::Int(now())),
("updatedAt", Value::Int(now())),
]);
put(db, &format!("item/{id}"), &item)?;
put(
db,
&format!("item-by-project/{project}/{item_sequence:020}/{id}"),
&item,
)?;
string(&id)
}
"archived" => {
archived += 1;
let archive = object(vec![
("id", string(&entity_id)),
("project", string(project)),
("source", string(source)),
("sourceKey", string(source_key)),
(
"title",
string(field(entity, "title").unwrap_or(source_key)),
),
("state", string(field(entity, "state").unwrap_or("Unknown"))),
("rawRef", string(&raw_ref)),
("createdAt", Value::Int(now())),
]);
put(
db,
&format!("archive-entity/{source_id}/{entity_id}"),
&archive,
)?;
string(&entity_id)
}
_ => {
quarantined += 1;
Value::Null
}
};
let disposition = object(vec![
("source", string(source)),
("sourceKey", string(source_key)),
("outcome", string(outcome)),
("destination", destination),
(
"rawRef",
if field(entity, "rawBase64").is_some() {
string(&raw_ref)
} else {
Value::Null
},
),
(
"diagnostic",
entity.get("diagnostic").cloned().unwrap_or(Value::Null),
),
]);
put(
db,
&format!("import-disposition/{run_id}/{entity_id}"),
&disposition,
)?;
}
let run = object(vec![
("id", string(run_id)),
("source", string(source)),
("project", string(project)),
("role", string(role)),
("planDigest", string(digest)),
("entities", Value::Int(entities.len() as i64)),
("imported", Value::Int(imported)),
("archived", Value::Int(archived)),
("quarantined", Value::Int(quarantined)),
("verified", Value::Bool(false)),
("createdAt", Value::Int(now())),
]);
put(db, &format!("import-run/{run_id}"), &run)?;
Ok(run)
})
}
pub fn verify_import(&self, run_id: &str) -> Result<Value> {
let mut db = self
.db
.lock()
.map_err(|_| Error::Storage("lock poisoned".into()))?;
let run = get_value(&mut db, &format!("import-run/{run_id}"))?
.ok_or_else(|| Error::NotFound(format!("import run {run_id}")))?;
let expected = run.get("entities").and_then(Value::as_i64).unwrap_or(-1);
let prefix = format!("import-disposition/{run_id}/");
let dispositions = db
.range(prefix.as_bytes(), &upper_bound(prefix.as_bytes()))
.map_err(storage)?;
if dispositions.len() as i64 != expected {
return Err(Error::Conflict(format!(
"import disposition mismatch: expected {expected}, got {}",
dispositions.len()
)));
}
let mut raw_checked = 0i64;
for (_, bytes) in dispositions {
let disposition = decode(&bytes)?;
if let Some(raw_ref) = field(&disposition, "rawRef") {
let raw = get_value(&mut db, raw_ref)?
.ok_or_else(|| Error::Storage(format!("missing raw record {raw_ref}")))?;
let encoded = field(&raw, "base64")
.ok_or_else(|| Error::Storage("raw base64 missing".into()))?;
let decoded = base64_decode(encoded)?;
if decoded.len() as i64 != raw.get("bytes").and_then(Value::as_i64).unwrap_or(-1)
|| bytes_digest(&decoded) != field(&raw, "sha256").unwrap_or("")
{
return Err(Error::Conflict(
"raw import digest verification failed".into(),
));
}
raw_checked += 1;
}
}
Ok(object(vec![
("ok", Value::Bool(true)),
("run", string(run_id)),
("dispositions", Value::Int(expected)),
("rawChecked", Value::Int(raw_checked)),
]))
}
}
fn discover(path: &Path, boards: &mut Vec<String>) -> Result<()> {
if !path.is_dir() {
return Ok(());
}
if path.file_name().and_then(|v| v.to_str()) == Some(".kanban") {
boards.push(
fs::canonicalize(path)
.map_err(storage)?
.display()
.to_string(),
);
return Ok(());
}
let name = path.file_name().and_then(|v| v.to_str()).unwrap_or("");
if matches!(name, ".git" | "target" | "node_modules" | ".cache") {
return Ok(());
}
for entry in fs::read_dir(path).map_err(storage)? {
discover(&entry.map_err(storage)?.path(), boards)?;
}
Ok(())
}
fn collect_cards(path: &Path, files: &mut Vec<PathBuf>) -> Result<()> {
if path.is_file() {
if path.extension().and_then(|v| v.to_str()) == Some("md")
&& path.file_name().and_then(|v| v.to_str()) != Some("README.md")
{
files.push(path.to_path_buf());
}
return Ok(());
}
for entry in fs::read_dir(path).map_err(storage)? {
let path = entry.map_err(storage)?.path();
let component = path.file_name().and_then(|v| v.to_str()).unwrap_or("");
if matches!(
component,
"policies" | "prompts" | "decisions" | "handoffs" | "events"
) {
continue;
}
collect_cards(&path, files)?;
}
Ok(())
}
fn state_from_path(path: &str) -> String {
for state in STATES {
let slug = state.to_ascii_lowercase().replace(' ', "-");
if path.to_ascii_lowercase().split('/').any(|part| {
part.trim_start_matches(|c: char| c.is_ascii_digit() || c == '-')
.starts_with(&slug)
}) {
return (*state).to_string();
}
}
"Inbox".into()
}
fn contains_secret(bytes: &[u8]) -> bool {
let text = String::from_utf8_lossy(bytes).to_ascii_lowercase();
[
"-----begin private key",
"password=",
"client_secret=",
"api_key=",
"authorization: bearer ",
]
.iter()
.any(|needle| text.contains(needle))
}
fn bytes_digest(bytes: &[u8]) -> String {
let digest = Sha256::digest(bytes);
let mut out = String::with_capacity(64);
for byte in digest {
out.push_str(&format!("{byte:02x}"));
}
out
}
fn base64(bytes: &[u8]) -> String {
const TABLE: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4);
for chunk in bytes.chunks(3) {
let value = ((chunk[0] as u32) << 16)
| ((chunk.get(1).copied().unwrap_or(0) as u32) << 8)
| chunk.get(2).copied().unwrap_or(0) as u32;
out.push(TABLE[((value >> 18) & 63) as usize] as char);
out.push(TABLE[((value >> 12) & 63) as usize] as char);
out.push(if chunk.len() > 1 {
TABLE[((value >> 6) & 63) as usize] as char
} else {
'='
});
out.push(if chunk.len() > 2 {
TABLE[(value & 63) as usize] as char
} else {
'='
});
}
out
}
fn base64_decode(value: &str) -> Result<Vec<u8>> {
fn digit(byte: u8) -> Option<u8> {
match byte {
b'A'..=b'Z' => Some(byte - b'A'),
b'a'..=b'z' => Some(byte - b'a' + 26),
b'0'..=b'9' => Some(byte - b'0' + 52),
b'+' => Some(62),
b'/' => Some(63),
_ => None,
}
}
if !value.len().is_multiple_of(4) {
return Err(Error::Invalid("invalid raw base64".into()));
}
let mut out = Vec::with_capacity(value.len() / 4 * 3);
for chunk in value.as_bytes().chunks(4) {
let a = digit(chunk[0]).ok_or_else(|| Error::Invalid("invalid raw base64".into()))? as u32;
let b = digit(chunk[1]).ok_or_else(|| Error::Invalid("invalid raw base64".into()))? as u32;
let c = if chunk[2] == b'=' {
0
} else {
digit(chunk[2]).ok_or_else(|| Error::Invalid("invalid raw base64".into()))? as u32
};
let d = if chunk[3] == b'=' {
0
} else {
digit(chunk[3]).ok_or_else(|| Error::Invalid("invalid raw base64".into()))? as u32
};
let n = (a << 18) | (b << 12) | (c << 6) | d;
out.push((n >> 16) as u8);
if chunk[2] != b'=' {
out.push((n >> 8) as u8);
}
if chunk[3] != b'=' {
out.push(n as u8);
}
}
Ok(out)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn import_preserves_raw_bytes_and_quarantines_secrets() {
let base = std::env::temp_dir().join(format!(
"akurai-tasks-import-{}-{}",
std::process::id(),
now()
));
let board = base.join(".kanban/columns/10-inbox");
fs::create_dir_all(&board).unwrap();
fs::write(board.join("TASK-1.md"), b"# First\nexact\r\nbytes\n").unwrap();
fs::write(board.join("TASK-2.md"), b"# Secret\npassword=bad\n").unwrap();
let bundle = plan_file_board(&base.join(".kanban"), "IMP", "live").unwrap();
assert_eq!(bundle.get("quarantined").and_then(Value::as_i64), Some(1));
let database = base.join("tasks.db");
let store = Store::open(&database).unwrap();
let run = store.apply_import(&bundle, "admin", "import").unwrap();
let verification = store.verify_import(field(&run, "id").unwrap()).unwrap();
assert_eq!(verification.get("ok"), Some(&Value::Bool(true)));
let count = match store.list_items("IMP").unwrap() {
Value::Array(values) => values.len(),
_ => 0,
};
assert_eq!(count, 1);
drop(store);
let _ = fs::remove_dir_all(base);
}
}