New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add custom threshold support for Nebula #115

Merged

DJAndries merged 7 commits into master from custom-threshold

Jun 20, 2024

Collaborator

DJAndries commented Mar 12, 2024

No description provided.

DJAndries force-pushed the custom-threshold branch 2 times, most recently from 26ae9ca to d4b46eb Compare

March 20, 2024 07:22

DJAndries changed the title ~~WIP: Add custom threshold support for Nebula~~ Add custom threshold support for Nebula

DJAndries marked this pull request as ready for review

March 20, 2024 19:16

DJAndries requested a review from a team as a code owner

March 20, 2024 19:16

rillian reviewed

View reviewed changes

misc/test-client/Cargo.toml Outdated Show resolved Hide resolved

rillian reviewed

View reviewed changes

misc/test-client/src/main.rs Show resolved Hide resolved

rillian approved these changes

View reviewed changes

Contributor

rillian left a comment

lgtm. So the idea here is to group submissions by their threshold value (stated or implicit) and decode and aggregate them separately, so the db will have separate entries for each threshold class and we can join them in the analysis dashboard queries?

misc/test-client/src/main.rs Outdated Show resolved Hide resolved

src/record_stream.rs

-                  let record: FutureRecord<str, [u8]> = FutureRecord::to(&self.topic).payload(record);
+                  let mut record: FutureRecord<str, [u8]> = FutureRecord::to(&self.topic).payload(record);
+                  if let Some(threshold) = request_threshold {
+                    let threshold = (threshold as u32).to_le_bytes();

Contributor

rillian Mar 20, 2024

I don't know anything about kafka, but it looks like this field really is just 'bytes' and it's custom serialization up to the application even for a base type like this. Is that right?

Collaborator Author

DJAndries Mar 20, 2024

correct, rust-rdkafka requires a &[u8] for header values, so we're responsible for serialization.

src/server.rs

Comment on lines +121 to +122

		if !state.request_threshold_range.contains(&threshold) {
		return Err(WebError::BadThreshold);

Contributor

rillian Mar 20, 2024

Ok, the server only accepts threshold values within whatever range we set (between 20 and 50 by default) so there's a bound to how many queues it will build. Excellent.

Collaborator Author

DJAndries commented Mar 20, 2024 •

edited

Loading

So the idea here is to group submissions by their threshold value (stated or implicit) and decode and aggregate them separately, so the db will have separate entries for each threshold class and we can join them in the analysis dashboard queries?

yes. pending messages (messages stored in PG from previous aggregation attempts) and new messages (messages that were just received via Kafka during the current iteration) will be grouped by tag, and then by threshold. New pending messages will be stored by tag and threshold.

However, recovered messages (which include the key, metric name and metric value) will continue to be grouped by tag. If there are multiple threshold values for a given tag (i.e. if we switched the Nebula k-threshold at some point), we're covered. It turns out that the key for a given tag remains the same, regardless of the multiple threshold values in the wild.

github-actions bot added the puLL-Merge label

github-actions bot reviewed

View reviewed changes

src/aggregator/processing.rs

+                      let msgs = if recovery_threshold == Some(threshold) && key_recovery_msgs.is_some() {
+                        // messages for this threshold were already drained in the key
+                        // recovery step, so use this existing vec
+                        key_recovery_msgs.take().unwrap()

github-actions bot Mar 21, 2024

_{reported by reviewdog 🐶}
[semgrep] expect or unwrap called in function returning a Result

Source: https://semgrep.dev/r/trailofbits.rs.panic-in-function-returning-result.panic-in-function-returning-result

Cc @thypon @bcaller

src/aggregator/processing.rs

@@ @@ -112,22 +198,15 @@ @@
                         id: 0,
                         msg_tag: msg_tag.clone(),
                         epoch_tag: *epoch as i16,
-                        metric_name: measurement.0,
-                        metric_value: measurement.1,
+                        metric_name: metric_name.unwrap(),

github-actions bot Mar 21, 2024

_{reported by reviewdog 🐶}
[semgrep] expect or unwrap called in function returning a Result

Source: https://semgrep.dev/r/trailofbits.rs.panic-in-function-returning-result.panic-in-function-returning-result

Cc @thypon @bcaller

src/aggregator/processing.rs

    
                        metric_name: measurement.0,

                        metric_value: measurement.1,

                        metric_name: metric_name.unwrap(),

                        metric_value: metric_value.unwrap(),

github-actions bot Mar 21, 2024

_{reported by reviewdog 🐶}
[semgrep] expect or unwrap called in function returning a Result

Source: https://semgrep.dev/r/trailofbits.rs.panic-in-function-returning-result.panic-in-function-returning-result

Cc @thypon @bcaller

src/aggregator/processing.rs

+                let key = if let Some(rec_msg) = existing_rec_msg {
+                  rec_msg.key.clone()
+                } else {
+                  let threshold = recovery_threshold.unwrap();

github-actions bot Mar 21, 2024

_{reported by reviewdog 🐶}
[semgrep] expect or unwrap called in function returning a Result

Source: https://semgrep.dev/r/trailofbits.rs.panic-in-function-returning-result.panic-in-function-returning-result

Cc @thypon @bcaller

src/aggregator/processing.rs

+                        AppSTARError::Recovery(ConstellationError::ShareRecovery) => {
+                          // Store new messages until we receive more shares in the future.
+                          for msg in msgs.drain(..new_msg_count) {
+                            chunk.new_msgs.get_mut(&threshold).unwrap().push(msg);

github-actions bot Mar 21, 2024

_{reported by reviewdog 🐶}
[semgrep] expect or unwrap called in function returning a Result

Source: https://semgrep.dev/r/trailofbits.rs.panic-in-function-returning-result.panic-in-function-returning-result

Cc @thypon @bcaller

github-actions bot added the needs-security-review label

github-actions bot assigned bcaller and thypon

Member

diracdeltas commented Apr 8, 2024

i see this adds new dependencies; is there something in the pre-merge pipeline which runs cargo audit? if not it should be added.

Member

thypon commented Apr 8, 2024

Is the modified code part of a command or a server? @DJAndries

Collaborator Author

DJAndries commented Apr 8, 2024

Is the modified code part of a command or a server? @DJAndries

modifications are mostly for the server + server-side aggregator. some changes were made to the test client as well

Contributor

rillian commented Apr 11, 2024

i see this adds new dependencies; is there something in the pre-merge pipeline which runs cargo audit? if not it should be added.

I filed #133 to address this. An unmaintained warning about the rusoto crates is the only thing reported after running cargo update.

DJAndries force-pushed the custom-threshold branch 2 times, most recently from 6878772 to e81c32d Compare

April 11, 2024 18:41

rillian mentioned this pull request

Merged

DJAndries added 6 commits

June 20, 2024 10:53


          Add request threshold header in server

7185d85


          Add multiple threshold support in aggregation process

d6a0885


          Add threshold header and randomness server request support for test c…

888efb2

…lient


          Separate message groups by threshold, and aggregate separately

7f91108


          Slight cleanup of processing function, ability to generate & send mea…

96967a3

…surements from csv using test client in one command


          Test client improvements as per feedback

4653dd8

DJAndries force-pushed the custom-threshold branch from 0ff661f to 4653dd8 Compare

June 20, 2024 18:00

github-actions bot commented Jun 20, 2024

[puLL-Merge] - brave/constellation-processors@115

Description

This PR makes changes to support custom k-threshold values in requests to the encrypted message submission endpoint. The k-threshold value is passed in a request header, and if present, overrides the default threshold. Pending messages are now stored in the database with the threshold value. Message grouping and key recovery is done separately for each threshold group.

Changes

Changes

migrations/: Adds a new migration to add a threshold column to the pending_msgs table
misc/test-client/:
- Adds support for submitting requests to a randomness server instead of using a local fetcher
- Adds a threshold header to requests based on a CLI argument
- Updates dependencies
src/aggregator/:
- consume.rs: Passes the default k-threshold value to message grouping
- group.rs:
  - Updates MessageChunk to store pending/new messages in a map keyed by threshold
  - Updates GroupedMessages.add() to use MessageWithThreshold
  - Updates pending message storage to include threshold
- mod.rs: Renames k-threshold constants
- processing.rs:
  - Updates layer processing to handle each threshold group separately
  - Removes k-threshold argument since it's no longer a single value
src/models/:
- Adds MessageWithThreshold struct
- Adds threshold field to PendingMessage
src/record_stream.rs:
- Adds support for threshold header in Kafka records
- Introduces ConsumedRecord to include the threshold value
src/server.rs:
- Adds support for brave-p3a-constellation-threshold header
- Validates the threshold header value is within configured range
- Passes threshold value when producing Kafka record

Security Hotspots

The brave-p3a-constellation-threshold header value is user controlled. The value is validated to be within a configured range to mitigate abuse.
The randomness server URL is user controlled in the test client. Ensure the randomness server is trusted.


          Fix new message count

1c5da4b

DJAndries merged commit 3283da2 into master

5 checks passed

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

needs-security-review puLL-Merge