The hub keeps Terraform and OpenTofu state (closes #252) 8dcaeeb848c6

-- Terraform and OpenTofu state kept by the hub (oxygenhub#252): one row

-- per repository and state name, its body sealed with the hub's secrets

-- key, its lock, and the last versions for rolling back.

CREATE TABLE terraform_states (

    repo_id    BIGINT NOT NULL REFERENCES repositories(id) ON DELETE CASCADE,

    name       TEXT NOT NULL,

    body       BYTEA NOT NULL,

    serial     BIGINT NOT NULL DEFAULT 0,

    lock_id    TEXT NOT NULL DEFAULT '',

    lock_info  TEXT NOT NULL DEFAULT '',

    locked_by  TEXT NOT NULL DEFAULT '',

    locked_ms  BIGINT,

    updated_by TEXT NOT NULL DEFAULT '',

    updated_ms BIGINT NOT NULL,

    PRIMARY KEY (repo_id, name)

);

CREATE TABLE terraform_state_versions (

    id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,

    repo_id    BIGINT NOT NULL REFERENCES repositories(id) ON DELETE CASCADE,

    name       TEXT NOT NULL,

    body       BYTEA NOT NULL,

    serial     BIGINT NOT NULL,

    created_by TEXT NOT NULL,

    created_ms BIGINT NOT NULL

);

CREATE INDEX terraform_state_versions_state ON terraform_state_versions(repo_id, name, id);

-- Terraform and OpenTofu state kept by the hub (oxygenhub#252): one row

-- per repository and state name, its body sealed with the hub's secrets

-- key, its lock, and the last versions for rolling back.

CREATE TABLE terraform_states (

    repo_id    INTEGER NOT NULL REFERENCES repositories(id) ON DELETE CASCADE,

    name       TEXT NOT NULL,

    body       BLOB NOT NULL,

    serial     INTEGER NOT NULL DEFAULT 0,

    lock_id    TEXT NOT NULL DEFAULT '',

    lock_info  TEXT NOT NULL DEFAULT '',

    locked_by  TEXT NOT NULL DEFAULT '',

    locked_ms  INTEGER,

    updated_by TEXT NOT NULL DEFAULT '',

    updated_ms INTEGER NOT NULL,

    PRIMARY KEY (repo_id, name)

);

CREATE TABLE terraform_state_versions (

    id         INTEGER PRIMARY KEY,

    repo_id    INTEGER NOT NULL REFERENCES repositories(id) ON DELETE CASCADE,

    name       TEXT NOT NULL,

    body       BLOB NOT NULL,

    serial     INTEGER NOT NULL,

    created_by TEXT NOT NULL,

    created_ms INTEGER NOT NULL

);

CREATE INDEX terraform_state_versions_state ON terraform_state_versions(repo_id, name, id);

//! Terraform and OpenTofu states (migration 0079, oxygenhub#252).

use super::{now_ms, Pool};

use sqlx::FromRow;

#[derive(Clone, Debug, FromRow)]

pub struct TfState {

    pub repo_id: i64,

    pub name: String,

    /// Sealed with the hub's secrets key; empty once deleted.

    pub body: Vec<u8>,

    pub serial: i64,

    pub lock_id: String,

    pub lock_info: String,

    pub locked_by: String,

    pub locked_ms: Option<i64>,

    pub updated_by: String,

    pub updated_ms: i64,

}

pub async fn get(pool: &Pool, repo_id: i64, name: &str) -> sqlx::Result<Option<TfState>> {

    sqlx::query_as("SELECT * FROM terraform_states WHERE repo_id = $1 AND name = $2").bind(repo_id).bind(name).fetch_optional(pool).await

}

pub async fn of_repo(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<TfState>> {

    sqlx::query_as("SELECT * FROM terraform_states WHERE repo_id = $1 ORDER BY name").bind(repo_id).fetch_all(pool).await

}

/// How many earlier versions a state keeps.

pub async fn versions(pool: &Pool, repo_id: i64, name: &str) -> sqlx::Result<i64> {

    sqlx::query_scalar("SELECT COUNT(*) FROM terraform_state_versions WHERE repo_id = $1 AND name = $2")

        .bind(repo_id)

        .bind(name)

        .fetch_one(pool)

        .await

}

/// Store a new body; the one it replaces becomes a version, and only the

/// last `keep` versions stay.

pub async fn put(pool: &Pool, repo_id: i64, name: &str, body: &[u8], serial: i64, by: &str, keep: i64) -> sqlx::Result<()> {

    let now = now_ms();

    let mut tx = pool.begin().await?;

    sqlx::query(

        "INSERT INTO terraform_state_versions (repo_id, name, body, serial, created_by, created_ms) \

         SELECT repo_id, name, body, serial, updated_by, updated_ms FROM terraform_states \

         WHERE repo_id = $1 AND name = $2 AND length(body) > 0",

    )

    .bind(repo_id)

    .bind(name)

    .execute(&mut *tx)

    .await?;

    sqlx::query(

        "INSERT INTO terraform_states (repo_id, name, body, serial, updated_by, updated_ms) VALUES ($1, $2, $3, $4, $5, $6) \

         ON CONFLICT (repo_id, name) DO UPDATE SET body = $3, serial = $4, updated_by = $5, updated_ms = $6",

    )

    .bind(repo_id)

    .bind(name)

    .bind(body)

    .bind(serial)

    .bind(by)

    .bind(now)

    .execute(&mut *tx)

    .await?;

    sqlx::query(

        "DELETE FROM terraform_state_versions WHERE repo_id = $1 AND name = $2 AND id NOT IN \

         (SELECT id FROM terraform_state_versions WHERE repo_id = $1 AND name = $2 ORDER BY id DESC LIMIT $3)",

    )

    .bind(repo_id)

    .bind(name)

    .bind(keep)

    .execute(&mut *tx)

    .await?;

    tx.commit().await

}

/// Delete a state: its body goes (its versions stay, for a mistake).

pub async fn delete(pool: &Pool, repo_id: i64, name: &str) -> sqlx::Result<()> {

    sqlx::query("DELETE FROM terraform_states WHERE repo_id = $1 AND name = $2").bind(repo_id).bind(name).execute(pool).await?;

    Ok(())

}

/// Take the lock; false when someone holds it.

pub async fn lock(pool: &Pool, repo_id: i64, name: &str, id: &str, info: &str, by: &str) -> sqlx::Result<bool> {

    let now = now_ms();

    let r = sqlx::query(

        "INSERT INTO terraform_states (repo_id, name, body, lock_id, lock_info, locked_by, locked_ms, updated_ms) \

         VALUES ($1, $2, $3, $4, $5, $6, $7, $7) \

         ON CONFLICT (repo_id, name) DO UPDATE SET lock_id = $4, lock_info = $5, locked_by = $6, locked_ms = $7 \

         WHERE terraform_states.lock_id = ''",

    )

    .bind(repo_id)

    .bind(name)

    .bind(Vec::<u8>::new())

    .bind(id)

    .bind(info)

    .bind(by)

    .bind(now)

    .execute(pool)

    .await?;

    Ok(r.rows_affected() == 1)

}

pub async fn unlock(pool: &Pool, repo_id: i64, name: &str) -> sqlx::Result<()> {

    sqlx::query(

        "UPDATE terraform_states SET lock_id = '', lock_info = '', locked_by = '', locked_ms = NULL WHERE repo_id = $1 AND name = $2",

    )

    .bind(repo_id)

    .bind(name)

    .execute(pool)

    .await?;

    Ok(())

}

//! Terraform and OpenTofu state, kept by the hub per repository

//! (oxygenhub#252): the HTTP backend Terraform and OpenTofu speak, with

//! locking, at `/api/v1/repos/OWNER/NAME/terraform/state/STATE`, so a Flow

//! job keeps its state without any cloud credential. A job authenticates

//! with an identity token of this hub's for the audience `{base}/terraform`

//! (its pipeline's `identity` block) or its runner's job token, either for

//! its own repository only; people with a token of theirs and their

//! repository role. Tokens go as the bearer, or as the password of Basic

//! auth, which is what the backend sends.

//!

//! State is kept outside the repository's history (it often holds secrets),

//! sealed at rest with the hub's secrets key, and its last versions are

//! kept, so a bad apply can be rolled back.

use crate::api::ApiError;

use crate::auth::roles::{Access, Role};

use crate::{db, Shared};

use axum::body::Bytes;

use axum::extract::{Path, Query, State};

use axum::http::{header, HeaderMap, Method, StatusCode};

use axum::response::{IntoResponse, Response};

/// Versions of a state kept besides the current one.

pub const VERSIONS_KEPT: i64 = 20;

/// The most a state may weigh.

pub const MAX_STATE_BYTES: usize = 32 << 20;

/// Who may use a repository's states.

enum Who {

    /// A running job of this repository.

    Job { name: String },

    /// A person, with their role on the repository.

    Person { account: Box<db::accounts::Account>, role: Role },

}

impl Who {

    fn label(&self) -> String {

        match self {

            Who::Job { name } => name.clone(),

            Who::Person { account, .. } => account.username.clone(),

        }

    }

    fn may(&self, access: Access) -> bool {

        match self {

            // A job reads and writes its own repository's states.

            Who::Job { .. } => true,

            Who::Person { role, .. } => crate::auth::roles::allows(Some(*role), access),

        }

    }

}

fn err(status: StatusCode, why: impl Into<String>) -> ApiError {

    ApiError(status, why.into())

}

/// The token a request carries: the bearer, or Basic auth's password.

fn token_of(headers: &HeaderMap) -> Option<String> {

    if let Some(t) = oxy_server::bearer_token(headers) {

        return Some(t.to_string());

    }

    let v = headers.get(header::AUTHORIZATION)?.to_str().ok()?;

    let b64 = v.strip_prefix("Basic ").or_else(|| v.strip_prefix("basic "))?;

    let raw = String::from_utf8(oxy_core::dsse::base64_decode(b64.trim())?).ok()?;

    let (_, password) = raw.split_once(':')?;

    (!password.is_empty()).then(|| password.to_string())

}

/// The audience a job's identity token needs here.

pub fn audience(base_url: &str) -> String {

    format!("{}/terraform", base_url.trim_end_matches('/'))

}

fn valid_name(name: &str) -> bool {

    !name.is_empty()

        && name.len() <= 100

        && name.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.')

        && !name.starts_with('.')

}

async fn who(app: &Shared, headers: &HeaderMap, row: &db::repos::RepoRow) -> Result<Who, ApiError> {

    let token = token_of(headers).ok_or_else(|| {

        err(

            StatusCode::UNAUTHORIZED,

            "a Terraform state needs a token: a job's (TF_HTTP_PASSWORD), or one of yours with repo:read or repo:write",

        )

    })?;

    let hash = crate::auth::token_hash(&token);

    if token.starts_with(crate::flow::JOB_TOKEN_PREFIX) {

        let job = db::flow::job_by_token(&app.db, &hash, crate::flow::JOB_TOKEN_MAX_AGE_MS)

            .await

            .map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?

            .ok_or_else(|| err(StatusCode::UNAUTHORIZED, "this job token is not valid (the job must be running)"))?;

        let run = db::flow::run_by_id(&app.db, job.run_id)

            .await

            .map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?

            .ok_or_else(|| err(StatusCode::UNAUTHORIZED, "this job token is not valid"))?;

        if run.repo_id != row.id {

            return Err(err(StatusCode::FORBIDDEN, "a job keeps the states of its own repository only"));

        }

        return Ok(Who::Job { name: format!("run #{} {}", run.number, job.name) });

    }

    let personal = db::accounts::token_by_hash(&app.db, &hash).await.map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    let Some(token) = personal.filter(|t| !t.expired(db::now_ms())) else {

        // A job's identity token for this hub's Terraform audience.

        let issuer = app.oidc().await.map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

        let aud = audience(app.config.base_url());

        let Some(claims) = issuer.verify(&token, &aud, db::now_ms()) else {

            return Err(err(StatusCode::UNAUTHORIZED, format!("unknown or expired token (a job's identity token must be for {aud})")));

        };

        let sub = claims["sub"].as_str().unwrap_or("");

        let repo = claims["repository"]

            .as_str()

            .map(String::from)

            .or_else(|| sub.strip_prefix("repo:").and_then(|r| r.split_once(":bookmark:")).map(|(r, _)| r.to_string()));

        if repo.as_deref() != Some(row.slug().as_str()) {

            return Err(err(StatusCode::FORBIDDEN, "a job keeps the states of its own repository only"));

        }

        return Ok(Who::Job { name: sub.to_string() });

    };

    if !(token.has_scope("repo:read") || token.has_scope("repo:write") || token.has_scope("repo:admin")) {

        return Err(err(StatusCode::FORBIDDEN, "the token lacks repo:read (to read a state) or repo:write (to change one)"));

    }

    let slug = row.slug();

    if !token.covers(&slug) {

        return Err(err(StatusCode::FORBIDDEN, format!("the token does not cover {slug}")));

    }

    let account = db::accounts::account_by_id(&app.db, token.account_id)

        .await

        .map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?

        .filter(|a| !a.banned && !a.disabled)

        .ok_or_else(|| err(StatusCode::UNAUTHORIZED, "unknown token"))?;

    let role = crate::auth::roles::role_for(&app.db, Some(&account), row)

        .await

        .map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    // A read-only token reads, whatever the role.

    let role = match role {

        Some(r) if !(token.has_scope("repo:write") || token.has_scope("repo:admin")) => Some(r.min(Role::Read)),

        r => r,

    };

    match role {

        Some(r) if crate::auth::roles::allows(Some(r), Access::Read) => Ok(Who::Person { account: Box::new(account), role: r }),

        _ => Err(err(StatusCode::NOT_FOUND, "no such repository")),

    }

}

#[derive(serde::Deserialize, Default)]

pub struct LockQuery {

    #[serde(rename = "ID", default)]

    id: Option<String>,

}

fn json(status: StatusCode, body: String) -> Response {

    (status, [(header::CONTENT_TYPE, "application/json")], body).into_response()

}

/// The lock's ID in a lock-info body (`{"ID": …}`).

fn lock_id(info: &[u8]) -> Option<String> {

    serde_json::from_slice::<serde_json::Value>(info).ok()?.get("ID")?.as_str().map(String::from)

}

/// `GET|POST|DELETE|LOCK|UNLOCK /api/v1/repos/{owner}/{name}/terraform/state/{state}`.

pub async fn state(

    State(app): State<Shared>,

    method: Method,

    headers: HeaderMap,

    Path((owner, name, state)): Path<(String, String, String)>,

    Query(q): Query<LockQuery>,

    body: Bytes,

) -> Result<Response, ApiError> {

    if !valid_name(&state) {

        return Err(err(StatusCode::BAD_REQUEST, "a state's name is letters, digits, '-', '_' and '.', up to 100"));

    }

    let row = db::repos::get(&app.db, &owner, &name)

        .await

        .map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?

        .ok_or_else(|| err(StatusCode::NOT_FOUND, "no such repository"))?;

    let who = who(&app, &headers, &row).await?;

    let write = method != Method::GET;

    if !who.may(if write { Access::Write } else { Access::Read }) {

        return Err(err(StatusCode::FORBIDDEN, "changing a state needs the write role"));

    }

    let db_err = |e: sqlx::Error| err(StatusCode::INTERNAL_SERVER_ERROR, e.to_string());

    let current = db::terraform::get(&app.db, row.id, &state).await.map_err(db_err)?;

    let locked = current.as_ref().filter(|s| !s.lock_id.is_empty());

    match method.as_str() {

        "GET" => match current.filter(|s| !s.body.is_empty()) {

            Some(s) => {

                let plain =

                    app.secret_key.open(&s.body).ok_or_else(|| err(StatusCode::INTERNAL_SERVER_ERROR, "the state does not open"))?;

                Ok(json(StatusCode::OK, plain))

            }

            // No state yet: Terraform starts afresh.

            None => Ok(StatusCode::NO_CONTENT.into_response()),

        },

        "POST" => {

            if body.len() > MAX_STATE_BYTES {

                return Err(err(StatusCode::PAYLOAD_TOO_LARGE, format!("a state may be {} MiB at most", MAX_STATE_BYTES >> 20)));

            }

            let text = std::str::from_utf8(&body).map_err(|_| err(StatusCode::BAD_REQUEST, "a state is JSON"))?;

            let parsed: serde_json::Value = serde_json::from_str(text).map_err(|_| err(StatusCode::BAD_REQUEST, "a state is JSON"))?;

            if let Some(l) = locked {

                if q.id.as_deref() != Some(l.lock_id.as_str()) {

                    return Ok(json(StatusCode::LOCKED, l.lock_info.clone()));

                }

            }

            let serial = parsed.get("serial").and_then(|v| v.as_i64()).unwrap_or(0);

            let sealed = app.secret_key.seal(text);

            db::terraform::put(&app.db, row.id, &state, &sealed, serial, &who.label(), VERSIONS_KEPT).await.map_err(db_err)?;

            Ok(StatusCode::OK.into_response())

        }

        "DELETE" => {

            if let Some(l) = locked {

                if q.id.as_deref() != Some(l.lock_id.as_str()) {

                    return Ok(json(StatusCode::LOCKED, l.lock_info.clone()));

                }

            }

            db::terraform::delete(&app.db, row.id, &state).await.map_err(db_err)?;

            Ok(StatusCode::OK.into_response())

        }

        "LOCK" => {

            let info = std::str::from_utf8(&body).unwrap_or("");

            let Some(id) = lock_id(body.as_ref()) else {

                return Err(err(StatusCode::BAD_REQUEST, "a lock request is the lock's JSON, with its ID"));

            };

            if let Some(l) = locked {

                return Ok(json(if l.lock_id == id { StatusCode::OK } else { StatusCode::LOCKED }, l.lock_info.clone()));

            }

            if db::terraform::lock(&app.db, row.id, &state, &id, info, &who.label()).await.map_err(db_err)? {

                Ok(json(StatusCode::OK, info.to_string()))

            } else {

                // Taken meanwhile.

                let now = db::terraform::get(&app.db, row.id, &state).await.map_err(db_err)?;

                Ok(json(StatusCode::LOCKED, now.map(|s| s.lock_info).unwrap_or_default()))

            }

        }

        "UNLOCK" => {

            let Some(l) = locked else { return Ok(StatusCode::OK.into_response()) };

            // `terraform force-unlock` sends the ID; a plain unlock its lock info.

            let id = lock_id(body.as_ref()).or(q.id.clone());

            if id.as_deref() != Some(l.lock_id.as_str()) {

                return Ok(json(StatusCode::CONFLICT, l.lock_info.clone()));

            }

            db::terraform::unlock(&app.db, row.id, &state).await.map_err(db_err)?;

            Ok(StatusCode::OK.into_response())

        }

        other => {

            Err(err(StatusCode::METHOD_NOT_ALLOWED, format!("{other} is not a Terraform backend method (GET, POST, DELETE, LOCK, UNLOCK)")))

        }

    }

}

        .route(

            "/repos/{owner}/{name}/terraform/state/{state}",

            axum::routing::any(crate::terraform::state)

                .layer(axum::extract::DefaultBodyLimit::max(crate::terraform::MAX_STATE_BYTES + 4096)),

        )

pub mod terraform;

pub mod terraform;

    /// Terraform states the hub keeps for it, with their kept versions (#252).

    tf_states: Vec<(db::terraform::TfState, i64)>,

    let mut tf_states = Vec::new();

    for s in db::terraform::of_repo(&app.db, row.id).await.map_err(|e| Error::from_db(shell.clone(), e))? {

        let n = db::terraform::versions(&app.db, row.id, &s.name).await.map_err(|e| Error::from_db(shell.clone(), e))?;

        tf_states.push((s, n));

    }

        tf_states,

/// Terraform state (oxygenhub#252): the HTTP backend's calls as

/// Terraform and OpenTofu make them, with a running job's token as the

/// Basic-auth password; the lock is honoured, versions are kept, the state

/// is sealed at rest, and another repository's job is refused.

#[tokio::test(flavor = "multi_thread")]

async fn a_job_keeps_its_terraform_state_on_the_hub() {

    let h = hub().await;

    create(&h.app, "andy", "infra").await;

    create(&h.app, "andy", "other").await;

    let work = tempfile::tempdir().unwrap();

    let c = client(&format!("{}/andy/infra", h.base), work.path());

    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));

    write(work.path(), "test.sh", "echo ok\n");

    c.snapshot("test").unwrap();

    c.commit("Add a pipeline", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    assert_eq!(runs_of(&h.app, "andy", "infra").await.len(), 1);

    let job = claim(&h.app).await.expect("a job");

    let token = job["token"].as_str().unwrap().to_string();

    let basic = format!("Basic {}", oxy_core::dsse::base64(format!("flow:{token}").as_bytes()));

    let path = "/api/v1/repos/andy/infra/terraform/state/prod";

    let send = |method: &str, uri: String, auth: Option<String>, body: &str| {

        let mut req = Request::builder().method(method).uri(uri);

        if let Some(a) = auth {

            req = req.header(header::AUTHORIZATION, a);

        }

        let req = req.body(Body::from(body.to_string())).unwrap();

        let app = h.app.clone();

        async move {

            let res = oxygenhub::router(app).oneshot(req).await.unwrap();

            let st = res.status();

            let b = res.into_body().collect().await.unwrap().to_bytes();

            (st, String::from_utf8_lossy(&b).into_owned())

        }

    };

    let lock = r#"{"ID":"lock-1","Operation":"OperationTypeApply","Who":"flow@runner","Version":"1.8.0"}"#;

    // No state yet; no token, nothing.

    assert_eq!(send("GET", path.into(), None, "").await.0, StatusCode::UNAUTHORIZED);

    assert_eq!(send("GET", path.into(), Some(basic.clone()), "").await.0, StatusCode::NO_CONTENT);

    // Lock, write under the lock, unlock.

    let (st, body) = send("LOCK", path.into(), Some(basic.clone()), lock).await;

    assert_eq!(st, StatusCode::OK, "{body}");

    let other_lock = r#"{"ID":"lock-2","Operation":"OperationTypePlan"}"#;

    let (st, held) = send("LOCK", path.into(), Some(basic.clone()), other_lock).await;

    assert_eq!(st, StatusCode::LOCKED);

    assert!(held.contains("lock-1"), "the holder's lock info: {held}");

    let state1 = r#"{"version":4,"serial":1,"lineage":"l","resources":[{"secret":"hunter2"}]}"#;

    assert_eq!(send("POST", path.into(), Some(basic.clone()), state1).await.0, StatusCode::LOCKED, "not under its lock");

    assert_eq!(send("POST", format!("{path}?ID=lock-1"), Some(basic.clone()), state1).await.0, StatusCode::OK);

    assert_eq!(send("UNLOCK", path.into(), Some(basic.clone()), other_lock).await.0, StatusCode::CONFLICT, "not its lock");

    assert_eq!(send("UNLOCK", path.into(), Some(basic.clone()), lock).await.0, StatusCode::OK);

    let (st, got) = send("GET", path.into(), Some(basic.clone()), "").await;

    assert_eq!((st, got.as_str()), (StatusCode::OK, state1));

    // A second write keeps the first as a version; the body is sealed.

    let state2 = state1.replace("\"serial\":1", "\"serial\":2");

    assert_eq!(send("POST", path.into(), Some(basic.clone()), &state2).await.0, StatusCode::OK);

    let row = db::repos::get(&h.app.db, "andy", "infra").await.unwrap().unwrap();

    assert_eq!(db::terraform::versions(&h.app.db, row.id, "prod").await.unwrap(), 1);

    let kept = db::terraform::get(&h.app.db, row.id, "prod").await.unwrap().unwrap();

    assert_eq!(kept.serial, 2);

    assert!(!String::from_utf8_lossy(&kept.body).contains("hunter2"), "sealed at rest");

    // Another repository's state is not this job's; a bad name is refused.

    let (st, _) = send("GET", "/api/v1/repos/andy/other/terraform/state/prod".into(), Some(basic.clone()), "").await;

    assert_eq!(st, StatusCode::FORBIDDEN);

    assert_eq!(

        send("GET", "/api/v1/repos/andy/infra/terraform/state/..x".into(), Some(basic.clone()), "").await.0,

        StatusCode::BAD_REQUEST

    );

    // The site token is no person's: refused.

    let (st, _) = send("GET", path.into(), Some("Bearer site".into()), "").await;

    assert_eq!(st, StatusCode::UNAUTHORIZED, "the site token is not a person's token");

    // A job's identity token for the Terraform audience, as its pipeline's

    // identity block mounts it: its own repository only, that audience only.

    let issuer = h.app.oidc().await.unwrap();

    let now = db::now_ms();

    let aud = oxygenhub::terraform::audience(h.app.config.base_url());

    let own = issuer.mint_for(h.app.config.base_url(), "repo:andy/infra:bookmark:main", &aud, now, 600);

    let (st, got) = send("GET", path.into(), Some(format!("Bearer {own}")), "").await;

    assert_eq!((st, got.contains("\"serial\":2")), (StatusCode::OK, true));

    let other = issuer.mint_for(h.app.config.base_url(), "repo:andy/other:bookmark:main", &aud, now, 600);

    assert_eq!(send("GET", path.into(), Some(format!("Bearer {other}")), "").await.0, StatusCode::FORBIDDEN);

    let elsewhere = issuer.mint_for(h.app.config.base_url(), "repo:andy/infra:bookmark:main", "sts.amazonaws.com", now, 600);

    assert_eq!(send("GET", path.into(), Some(format!("Bearer {elsewhere}")), "").await.0, StatusCode::UNAUTHORIZED);

}

/// The real OpenTofu against the backend (oxygenhub#252): init, apply,

/// a second apply that changes nothing, all through the hub's state with a

/// job's token. Needs Docker and the `oxygenflow-tofu` image; run with

/// `cargo test -p oxygenhub --test flow -- --ignored tofu_keeps`.

#[tokio::test(flavor = "multi_thread")]

#[ignore]

async fn tofu_keeps_its_state_on_the_hub() {

    let h = hub().await;

    create(&h.app, "andy", "infra").await;

    let work = tempfile::tempdir().unwrap();

    let c = client(&format!("{}/andy/infra", h.base), work.path());

    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));

    write(work.path(), "test.sh", "echo ok\n");

    c.snapshot("test").unwrap();

    c.commit("Add a pipeline", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    assert_eq!(runs_of(&h.app, "andy", "infra").await.len(), 1);

    let job = claim(&h.app).await.expect("a job");

    let token = job["token"].as_str().unwrap().to_string();

    let tf = tempfile::tempdir().unwrap();

    let address = format!("{}/api/v1/repos/andy/infra/terraform/state/prod", h.base);

    std::fs::write(

        tf.path().join("main.tf"),

        format!(

            "terraform {{\n  backend \"http\" {{\n    address = \"{address}\"\n    lock_address = \"{address}\"\n    unlock_address = \"{address}\"\n    username = \"flow\"\n  }}\n}}\nresource \"terraform_data\" \"x\" {{\n  input = \"hello\"\n}}\n"

        ),

    )

    .unwrap();

    let tofu = |args: &[&str]| {

        let out = std::process::Command::new("docker")

            .args(["run", "--rm", "--network", "host", "-v"])

            .arg(format!("{}:/w", tf.path().display()))

            .args(["-w", "/w", "-e"])

            .arg(format!("TF_HTTP_PASSWORD={token}"))

            .args(["--entrypoint", "tofu", "oxygenflow-tofu:latest"])

            .args(args)

            .output()

            .expect("docker");

        let text = format!("{}{}", String::from_utf8_lossy(&out.stdout), String::from_utf8_lossy(&out.stderr));

        assert!(out.status.success(), "tofu {args:?}: {text}");

        text

    };

    tofu(&["init", "-input=false", "-no-color"]);

    let applied = tofu(&["apply", "-auto-approve", "-input=false", "-no-color"]);

    assert!(applied.contains("1 added"), "{applied}");

    let again = tofu(&["apply", "-auto-approve", "-input=false", "-no-color"]);

    assert!(again.contains("No changes"), "the state was kept: {again}");

    let row = db::repos::get(&h.app.db, "andy", "infra").await.unwrap().unwrap();

    let kept = db::terraform::get(&h.app.db, row.id, "prod").await.unwrap().unwrap();

    assert!(kept.lock_id.is_empty(), "unlocked after each run");

    assert!(kept.serial >= 1);

}

### 13.5.4 Terraform state (oxygenhub#252, built 2026-10-18)

The hub keeps Terraform and OpenTofu state per repository, so a pipeline needs no bucket or cloud credential just to hold it. It speaks their HTTP backend at `/api/v1/repos/OWNER/NAME/terraform/state/STATE` (STATE: letters, digits, `-`, `_`, `.`, up to 100):

- **GET** the state (204 when there is none);

- **POST** a new one: JSON, up to 32 MiB, with `?ID=` the lock's ID while it is locked;

- **DELETE** it;

- **LOCK** with the lock's JSON: 423 and the holder's lock when someone else holds it, 200 to its holder again;

- **UNLOCK** with the lock's JSON or `?ID=`: 409 for another lock's.

**Who may use it:**

- a job, with an identity token of this hub's for the audience `{base_url}/terraform` (its pipeline's `identity` block) or with its runner's job token, for its own repository only;

- a person, with a token of theirs (`repo:read` to read, `repo:write` to change) and the repository role to match.

The token goes as the bearer or as Basic auth's password, which is what the backend sends (`TF_HTTP_PASSWORD`).

**Storage:** each state is sealed at rest with the hub's secrets key, and the last 20 versions before it are kept (migration 0079: `terraform_states`, `terraform_state_versions`). Backups carry the sealed bodies, which open only with this hub's key. A repository's Settings › Terraform state lists its states, serials, writers, locks and versions, with the backend block to use. Not for the hub's own infrastructure (oxygenhub-gcp), whose state cannot live on the hub it deploys.

<section class="panel" id="terraform">

  <h2>Terraform state</h2>

  <p class="small">The hub keeps this repository's Terraform and OpenTofu state, sealed at rest, with its last {{ crate::terraform::VERSIONS_KEPT }} versions. A pipeline's job uses it with an identity token of this hub's (no cloud credential): give the job <code>identity "{{ shell.base_url }}/terraform" {% raw %}{{% endraw %} mount-path "/run/secrets/tf" {% raw %}}{% endraw %}</code>, and <code>TF_HTTP_PASSWORD=$(cat /run/secrets/tf)</code> beside:</p>

  <pre class="snippet">terraform {

  backend "http" {

    address        = "{{ shell.base_url }}/api/v1/repos/{{ row.owner }}/{{ row.name }}/terraform/state/NAME"

    lock_address   = "{{ shell.base_url }}/api/v1/repos/{{ row.owner }}/{{ row.name }}/terraform/state/NAME"

    unlock_address = "{{ shell.base_url }}/api/v1/repos/{{ row.owner }}/{{ row.name }}/terraform/state/NAME"

    username       = "flow"

  }

}</pre>

  {% if tf_states.is_empty() %}<p class="mute small">No state yet.</p>{% else %}

  <div class="table"><table>

  <tr><th>State</th><th>Serial</th><th>Last written</th><th>Lock</th><th>Versions kept</th></tr>

  {% for (s, n) in tf_states %}<tr><td><strong>{{ s.name }}</strong></td><td>{{ s.serial }}</td><td class="small">{% if s.updated_by.is_empty() %}<span class="mute">never</span>{% else %}by {{ s.updated_by }} <time title="{{ s.updated_ms|when_i }}">{{ s.updated_ms|ago_i }}</time>{% endif %}</td><td class="small">{% if s.lock_id.is_empty() %}<span class="mute">free</span>{% else %}held by {{ s.locked_by }}{% if let Some(t) = s.locked_ms %} <time title="{{ t|when_i }}">{{ t|ago_i }}</time>{% endif %}{% endif %}</td><td>{{ n }}</td></tr>{% endfor %}

  </table></div>{% endif %}

</section>