Auto merge of #1681 - sgrif:sg-rate-limit-publish, r=jtgeibel

Add more aggressive rate limiting for publishing new crates

This is still incomplete, but the bulk of the code has been written so I figured I'd get some eyes on it. Right now this just panics instead of returning an error if the user is out of tokens. Still left to do are:

- The two ignored test cases
- Implementing the actual error type
- Per-user burst rate overrides
- cron job to restrict the table size and clean up stale buckets (I probably won't land this in the initial PR, our users table needs to grow by 2 orders of magnitude for this to really matter -- but I do want to land it as a followup PR since I haven't tested this with cases where now - last_update is greater than a month. It should work fine but I'd rather not have this run against poorly defined semantics)

I think the limit we'll probably set to start is 1 req/10s with a burst of 30. The error message will tell folks they can either wait for {time until next token} or email us to get the limit increased for them. This is limited per user instead of per ip since rotating your user is harder than rotating your IP. It's stored in the DB since this is only for publishing new crates, which is slow enough already that the DB load of rate limiting there shouldn't matter.

I needed to update to Rust 1.33 to get `Duration::as_millis` (note: the way we're using this feature causes UB if the rate limit is slower than 1 request per 292471208 years. I assume this is not a problem)
I needed to update to Diesel 1.4.2 to get a fix for diesel-rs/diesel#2017

The algorithm used is pretty much the standard token bucket algorithm. It's *slightly* different in how we set `tokens = max(0, tokens - 1) + tokens_to_add` instead of `tokens = max(0, tokens_to_add + 1)`. This is because the usual implementation checks available tokens before subtracting them (and thus never persists if there aren't enough tokens available). Since we're doing this in a single query, and we can *only* return the final, persisted value, we have to change the calculation slightly to make sure that a user who is out of tokens gets `1` back after the rate limit.

A side effect of all of this is that our token count is actually offset by 1. 0 means the user is not only out of tokens, but that we just tried to take a token and couldn't. 1 means an empty bucket, and a full bucket would technically be burst + 1. The alternative would be -1 meaning the user is actually out of tokens, but since we only ever refill the bucket when we're trying to take a token, we never actually persist a full bucket. I figured a range of 0...burst made more sense than -1..burst.

RustConfig

@@ -1 +1 @@
-VERSION=1.32.0
+VERSION=1.33.0

migrations/2019-03-18-233900_create_publish_limit_buckets/down.sql

@@ -0,0 +1 @@
+DROP TABLE publish_limit_buckets;

migrations/2019-03-18-233900_create_publish_limit_buckets/up.sql

@@ -0,0 +1,5 @@
+CREATE TABLE publish_limit_buckets(
+  user_id INTEGER PRIMARY KEY NOT NULL REFERENCES users,
+  tokens INTEGER NOT NULL,
+  last_refill TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);

migrations/2019-04-04-192902_create_publish_rate_overrides/down.sql

@@ -0,0 +1 @@
+DROP TABLE publish_rate_overrides;

migrations/2019-04-04-192902_create_publish_rate_overrides/up.sql

@@ -0,0 +1,4 @@
+CREATE TABLE publish_rate_overrides (
+  user_id INTEGER PRIMARY KEY REFERENCES users,
+  burst INTEGER NOT NULL
+);

src/bin/update-downloads.rs

@@ -107,7 +107,7 @@ mod test {
             name: "foo",
             ..Default::default()
         }
-        .create_or_update(conn, None, user_id)
+        .create_or_update(conn, None, user_id, None)
         .unwrap();
         let version = NewVersion::new(
             krate.id,

src/config.rs

@@ -1,3 +1,4 @@
+use crate::publish_rate_limit::PublishRateLimit;
 use crate::{env, uploaders::Uploader, Env, Replica};
 use std::path::PathBuf;
 use url::Url;
@@ -16,6 +17,7 @@ pub struct Config {
     pub max_unpack_size: u64,
     pub mirror: Replica,
     pub api_protocol: String,
+    pub publish_rate_limit: PublishRateLimit,
 }
 
 impl Default for Config {
@@ -132,6 +134,7 @@ impl Default for Config {
             max_unpack_size: 512 * 1024 * 1024, // 512 MB max when decompressed
             mirror,
             api_protocol,
+            publish_rate_limit: Default::default(),
         }
     }
 }

src/controllers/krate/publish.rs

@@ -84,7 +84,12 @@ pub fn publish(req: &mut dyn Request) -> CargoResult<Response> {
         };
 
         let license_file = new_crate.license_file.as_ref().map(|s| &**s);
-        let krate = persist.create_or_update(&conn, license_file, user.id)?;
+        let krate = persist.create_or_update(
+            &conn,
+            license_file,
+            user.id,
+            Some(&app.config.publish_rate_limit),
+        )?;
 
         let owners = krate.owners(&conn)?;
         if user.rights(req.app(), &owners)? < Rights::Publish {

src/lib.rs

@@ -37,8 +37,10 @@ pub mod email;
 pub mod git;
 pub mod github;
 pub mod middleware;
+mod publish_rate_limit;
 pub mod render;
 pub mod schema;
+mod test_util;
 pub mod uploaders;
 pub mod util;
 

src/models/category.rs

@@ -183,12 +183,11 @@ impl<'a> NewCategory<'a> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::test_util::pg_connection_no_transaction;
     use diesel::connection::SimpleConnection;
 
     fn pg_connection() -> PgConnection {
-        let database_url =
-            dotenv::var("TEST_DATABASE_URL").expect("TEST_DATABASE_URL must be set to run tests");
-        let conn = PgConnection::establish(&database_url).unwrap();
+        let conn = pg_connection_no_transaction();
         // These tests deadlock if run concurrently
         conn.batch_execute("BEGIN; LOCK categories IN ACCESS EXCLUSIVE MODE")
             .unwrap();

src/models/krate.rs

@@ -15,6 +15,7 @@ use crate::models::{
 use crate::views::{EncodableCrate, EncodableCrateLinks};
 
 use crate::models::helpers::with_count::*;
+use crate::publish_rate_limit::PublishRateLimit;
 use crate::schema::*;
 
 /// Hosts in this list are known to not be hosting documentation,
@@ -105,6 +106,7 @@ impl<'a> NewCrate<'a> {
         conn: &PgConnection,
         license_file: Option<&'a str>,
         uploader: i32,
+        rate_limit: Option<&PublishRateLimit>,
     ) -> CargoResult<Crate> {
         use diesel::update;
 
@@ -115,6 +117,9 @@ impl<'a> NewCrate<'a> {
             // To avoid race conditions, we try to insert
             // first so we know whether to add an owner
             if let Some(krate) = self.save_new_crate(conn, uploader)? {
+                if let Some(rate_limit) = rate_limit {
+                    rate_limit.check_rate_limit(uploader, conn)?;
+                }
                 return Ok(krate);
             }
 

src/models/user.rs

@@ -21,7 +21,7 @@ pub struct User {
     pub gh_id: i32,
 }
 
-#[derive(Insertable, Debug)]
+#[derive(Insertable, Debug, Default)]
 #[table_name = "users"]
 pub struct NewUser<'a> {
     pub gh_id: i32,