An organisation's storage policy 8ddebaa7e856

-- An organisation's storage policy (oxygenhub#96): whether every new

-- repository must live in a bucket, and the store URL prefixes allowed.

CREATE TABLE org_storage_policy (

    org_id         INTEGER PRIMARY KEY REFERENCES organisations(id) ON DELETE CASCADE,

    require_bucket INTEGER NOT NULL DEFAULT 0,

    -- One store URL prefix per line; empty allows any bucket.

    allowed        TEXT NOT NULL DEFAULT ''

);

- **An organisation's storage policy (#96).** An organisation's Settings › Storage can require every new repository to live in a bucket and list the buckets allowed, so its data stays where its own buckets are. The New repository form refuses anything else and says what is allowed.

/// An organisation's storage policy: every new repository in a bucket,

/// and only in stores under these URL prefixes when there are any. The

/// buckets an organisation lists are its own, so where they are (the

/// region, the provider, the country) is where its data is.

#[derive(Clone, Debug, Default, PartialEq, Eq)]

pub struct Policy {

    pub require_bucket: bool,

    pub allowed: Vec<String>,

}

impl Policy {

    /// Whether a new repository may be kept at `store` (`None`: this

    /// hub's disk); otherwise what to tell its creator.

    pub fn permits(&self, org: &str, store: Option<&str>) -> Result<(), String> {

        match store {

            None if self.require_bucket => Err(format!(

                "{org} keeps every new repository in a bucket: choose “my own bucket” under Storage{}",

                if self.allowed.is_empty() { String::new() } else { format!(" ({})", self.allowed.join(", ")) }

            )),

            Some(url) if !self.allowed.is_empty() && !self.allowed.iter().any(|p| url.trim().starts_with(p.as_str())) => {

                Err(format!("{org} allows only these stores: {}", self.allowed.join(", ")))

            }

            _ => Ok(()),

        }

    }

}

pub async fn policy(pool: &Pool, org_id: i64) -> sqlx::Result<Policy> {

    let row: Option<(i64, String)> =

        sqlx::query_as("SELECT require_bucket, allowed FROM org_storage_policy WHERE org_id = ?").bind(org_id).fetch_optional(pool).await?;

    Ok(row

        .map(|(r, a)| Policy {

            require_bucket: r != 0,

            allowed: a.lines().map(str::trim).filter(|l| !l.is_empty()).map(String::from).collect(),

        })

        .unwrap_or_default())

}

/// Set a policy from what a form gave: the allowed prefixes one per line

/// (or comma-separated), each a store URL of a bucket.

pub async fn set_policy(pool: &Pool, org_id: i64, require_bucket: bool, allowed: &str) -> Result<Policy, String> {

    let mut prefixes = Vec::new();

    for p in allowed.split([',', '\n']).map(str::trim).filter(|p| !p.is_empty()) {

        if p.contains('?') {

            return Err(format!("{p}: a prefix names a bucket and a path, with no settings"));

        }

        check_url(p).map_err(|e| format!("{p}: {e}"))?;

        prefixes.push(p.to_string());

    }

    sqlx::query(

        "INSERT INTO org_storage_policy (org_id, require_bucket, allowed) VALUES (?, ?, ?) \

         ON CONFLICT (org_id) DO UPDATE SET require_bucket = excluded.require_bucket, allowed = excluded.allowed",

    )

    .bind(org_id)

    .bind(require_bucket as i64)

    .bind(prefixes.join("\n"))

    .execute(pool)

    .await

    .map_err(|e| e.to_string())?;

    Ok(Policy { require_bucket, allowed: prefixes })

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn store_urls_a_hub_accepts() {

        assert!(check_url("s3://bucket/repos/x?region=eu-west-2").is_ok());

        assert!(check_url("gs://bucket/x").is_ok());

        assert!(check_url("az://account/container/x").is_ok());

        for (bad, why) in [

            ("s3://b/x?profile=prod", "?profile="),

            ("s3://b/x?region=a&endpoint=http://10.0.0.1", "?endpoint="),

            ("gs://b/x?credentials=/etc/key.json", "?credentials="),

            ("dir:///srv", "a store is"),

            ("file:///srv", "a store is"),

            ("az://account", "the container"),

        ] {

            let e = check_url(bad).unwrap_err();

            assert!(e.contains(why), "{bad}: {e}");

        }

    }

    #[test]

    fn a_policy_says_what_it_allows() {

        let open = Policy::default();

        assert!(open.permits("acme", None).is_ok());

        let strict = Policy { require_bucket: true, allowed: vec!["s3://acme-london/".into(), "gs://acme-eu/".into()] };

        let e = strict.permits("acme", None).unwrap_err();

        assert!(e.contains("acme keeps every new repository in a bucket") && e.contains("s3://acme-london/"), "{e}");

        assert!(strict.permits("acme", Some("s3://acme-london/repos/x?region=eu-west-2")).is_ok());

        let e = strict.permits("acme", Some("s3://acme-virginia/x")).unwrap_err();

        assert!(e.contains("allows only these stores"), "{e}");

    }

}

        .route("/orgs/{name}/settings/storage", post(orgs::storage))

    /// Where its new repositories may live (#96).

    policy: crate::storage::Policy,

    let policy = crate::storage::policy(&app.db, org.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;

        policy,

#[derive(Deserialize)]

pub struct PolicyForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    #[serde(default)]

    require_bucket: Option<String>,

    #[serde(default)]

    allowed: String,

}

/// `POST /orgs/{name}/settings/storage`: where the organisation's new

/// repositories may live (#96).

pub async fn storage(

    State(app): State<Shared>,

    viewer: Viewer,

    Path(name): Path<String>,

    Form(f): Form<PolicyForm>,

) -> Result<Page<OrgSettingsPage>, Error> {

    let shell = Shell::new(&app, &format!("/orgs/{name}/settings"), &viewer);

    let (me, org) = owned_org(&app, &shell, &viewer, &name).await?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    match crate::storage::set_policy(&app.db, org.id, f.require_bucket.is_some(), &f.allowed).await {

        Ok(p) => {

            let what = match (p.require_bucket, p.allowed.is_empty()) {

                (true, true) => "every new repository in a bucket".to_string(),

                (true, false) => format!("every new repository in {}", p.allowed.join(", ")),

                (false, true) => "any storage".to_string(),

                (false, false) => format!("buckets only in {}", p.allowed.join(", ")),

            };

            accounts::audit(&app.db, Some(&me), "org.storage_policy", &org.name, &what).await.ok();

            org_page(&app, shell, org, Some(format!("Storage policy saved: {what}.")), None).await

        }

        Err(e) => org_page(&app, shell, org, None, Some(format!("Storage policy: {e}."))).await,

    }

}

    // An organisation's storage policy (oxygenhub#96).

    if let Some(org) = accounts::org_by_name(&app.db, &name.owner).await.map_err(|e| Error::from_db(shell.clone(), e))? {

        let policy = crate::storage::policy(&app.db, org.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;

        let store = (f.storage == "bucket").then_some(f.bucket.store_url.as_str());

        if let Err(e) = policy.permits(&org.name, store) {

            return Ok(fail(&format!("{e}.")));

        }

    }

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

async fn an_organisation_can_require_its_own_buckets() {

    let h = hub().await;

    let me = andy(&h.app).await;

    let csrf = csrf_of(&send(&h.app, "GET", "/orgs/new", Some(&me), None).await.1);

    let (st, body, _) = send(&h.app, "POST", "/orgs/new", Some(&me), Some(&format!("_csrf={csrf}&name=acme&display_name=Acme"))).await;

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

    let settings = send(&h.app, "GET", "/orgs/acme/settings", Some(&me), None).await.1;

    assert!(settings.contains("id=\"storage\""));

    let csrf = csrf_of(&settings);

    let set = |allowed: &str| {

        let form = format!("_csrf={csrf}&require_bucket=1&allowed={}", enc(allowed));

        let app = h.app.clone();

        let me = me.clone();

        async move { send(&app, "POST", "/orgs/acme/settings/storage", Some(&me), Some(&form)).await.1 }

    };

    let page = set("s3://acme-london/repos/?region=eu-west-2").await;

    assert!(page.contains("a prefix names a bucket and a path, with no settings"), "{page}");

    let page = set("s3://acme-london/repos/\ngs://acme-eu/").await;

    assert!(page.contains("Storage policy saved: every new repository in s3://acme-london/repos/, gs://acme-eu/."), "{page}");

    let new = |fields: String| {

        let app = h.app.clone();

        let me = me.clone();

        async move {

            let csrf = csrf_of(&send(&app, "GET", "/new", Some(&me), None).await.1);

            send(&app, "POST", "/new", Some(&me), Some(&format!("_csrf={csrf}&owner=acme&{fields}"))).await.1

        }

    };

    let page = new("name=ondisk".into()).await;

    assert!(page.contains("acme keeps every new repository in a bucket"), "{page}");

    let page = new(format!("name=elsewhere&storage=bucket&store_url={}&access_key_id=a&secret_access_key=b", enc("s3://other/x"))).await;

    assert!(page.contains("acme allows only these stores: s3://acme-london/repos/, gs://acme-eu/"), "{page}");

    assert!(db::repos::get(&h.app.db, "acme", "ondisk").await.unwrap().is_none());

    assert!(db::repos::get(&h.app.db, "acme", "elsewhere").await.unwrap().is_none());

    // A user's own repositories are not the organisation's business.

    let (st, body) = new_repo(&h.app, &me, "name=mine").await;

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

}

- **Organisation policy:** Settings › Storage on an organisation can require every new repository to live in a bucket, and can list the store URL prefixes allowed (its own buckets, so its data stays where they are). The New repository form refuses anything else for that organisation, saying what is allowed. Repositories that exist are not moved. Changes are audited as `org.storage_policy`.

- **Not yet:** forking such a repository is refused (clone and push instead). Deleting one leaves its data in the bucket, which is the owner's. The Sync status and runs on references moved outside the hub are still to come.

  <a href="#storage">Storage</a>

<form class="card" id="storage" method="post" action="/orgs/{{ org.name }}/settings/storage">

  <input type="hidden" name="_csrf" value="{{ shell.csrf }}">

  <h2>Storage</h2>

  <p class="small mute">Where the organisation's new repositories may live. Repositories that exist stay where they are.</p>

  <label class="check"><input type="checkbox" name="require_bucket" value="1"{% if policy.require_bucket %} checked{% endif %}> Keep every new repository in a bucket, not on this server's disk</label>

  <label for="allowed">Buckets allowed, one store URL prefix per line (empty: any bucket)</label>

  <textarea id="allowed" name="allowed" rows="3" placeholder="s3://acme-london/repos/&#10;gs://acme-eu/" spellcheck="false">{{ policy.allowed.join("\n") }}</textarea>

  <p class="small mute">Listing the organisation's own buckets keeps its data where those buckets are: the provider, the region, the country.</p>

  <button type="submit">SAVE</button>

</form>