From 0cb6ffd6ed265ca72fbf40cf7a2c3c0c0cbf991d Mon Sep 17 00:00:00 2001 From: Mathias Date: Thu, 2 May 2024 16:56:57 +0200 Subject: [PATCH 1/3] Add CI test for AWS IoT Device Advisor --- .github/workflows/ci.yml | 13 +++--- mqttrust_core/examples/aws_device_advisor.rs | 12 +++-- mqttrust_core/examples/secrets/identity.pfx | Bin 3349 -> 2627 bytes mqttrust_core/src/eventloop.rs | 46 +++++++++++++++---- mqttrust_core/src/lib.rs | 3 +- mqttrust_core/src/state.rs | 23 +++++++--- {.github/scripts => scripts}/da_monitor.sh | 0 scripts/rotate_secrets.sh | 27 +++++++++++ 8 files changed, 96 insertions(+), 28 deletions(-) rename {.github/scripts => scripts}/da_monitor.sh (100%) create mode 100755 scripts/rotate_secrets.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d05767e..1526897 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -72,9 +72,8 @@ jobs: AWS_DEFAULT_REGION: ${{ secrets.MGMT_AWS_DEFAULT_REGION }} AWS_ACCESS_KEY_ID: ${{ secrets.MGMT_AWS_ACCESS_KEY_ID }} AWS_SECRET_ACCESS_KEY: ${{ secrets.MGMT_AWS_SECRET_ACCESS_KEY }} - SUITE_ID: greb3uy2wtq3 - THING_ARN: arn:aws:iot:eu-west-1:274906834921:thing/mqttrust - CERTIFICATE_ARN: arn:aws:iot:eu-west-1:274906834921:cert/e7280d8d316b58da3058037a2c1730d9eb15de50e96f4d47e54ea655266b76db + SUITE_ID: 1gaev57dq6i5 + THING_ARN: arn:aws:iot:eu-west-1:411974994697:thing/mqttrust steps: - name: Checkout uses: actions/checkout@v1 @@ -89,7 +88,7 @@ jobs: - name: Get AWS_HOSTNAME id: hostname run: | - hostname=$(aws iotdeviceadvisor get-endpoint --output text --query endpoint) + hostname=$(aws iotdeviceadvisor get-endpoint --thing-arn ${{ env.THING_ARN }} --output text --query endpoint) ret=$? echo "::set-output name=AWS_HOSTNAME::$hostname" exit $ret @@ -105,7 +104,7 @@ jobs: - name: Start test suite id: test_suite run: | - suite_id=$(aws iotdeviceadvisor start-suite-run --suite-definition-id ${{ env.SUITE_ID }} --suite-run-configuration "primaryDevice={thingArn=${{ env.THING_ARN }},certificateArn=${{ env.CERTIFICATE_ARN }}}" --output text --query suiteRunId) + suite_id=$(aws iotdeviceadvisor start-suite-run --suite-definition-id ${{ env.SUITE_ID }} --suite-run-configuration "primaryDevice={thingArn=${{ env.THING_ARN }}},parallelRun=true" --output text --query suiteRunId) ret=$? echo "::set-output name=SUITE_RUN_ID::$suite_id" exit $ret @@ -121,9 +120,9 @@ jobs: - name: Monitor test run run: | - chmod +x ./.github/scripts/da_monitor.sh + chmod +x ./scripts/da_monitor.sh echo ${{ env.SUITE_ID }} ${{ steps.test_suite.outputs.SUITE_RUN_ID }} ${{ steps.binary.outputs.PID }} - ./.github/scripts/da_monitor.sh ${{ env.SUITE_ID }} ${{ steps.test_suite.outputs.SUITE_RUN_ID }} ${{ steps.binary.outputs.PID }} + ./scripts/da_monitor.sh ${{ env.SUITE_ID }} ${{ steps.test_suite.outputs.SUITE_RUN_ID }} ${{ steps.binary.outputs.PID }} - name: Kill test binary process if: ${{ always() }} diff --git a/mqttrust_core/examples/aws_device_advisor.rs b/mqttrust_core/examples/aws_device_advisor.rs index 0d86e29..8720095 100644 --- a/mqttrust_core/examples/aws_device_advisor.rs +++ b/mqttrust_core/examples/aws_device_advisor.rs @@ -1,5 +1,6 @@ mod common; +use mqttrust::encoding::v4::{Connack, ConnectReturnCode}; use mqttrust::{Mqtt, QoS, SubscribeTopic}; use mqttrust_core::bbqueue::BBBuffer; use mqttrust_core::{EventLoop, MqttOptions, Notification}; @@ -44,10 +45,13 @@ fn main() { .spawn(move || loop { match nb::block!(mqtt_eventloop.connect(&mut network)) { Err(_) => continue, - Ok(true) => { + Ok(Some(Notification::ConnAck(Connack { + session_present, + code: ConnectReturnCode::Accepted, + }))) => { log::info!("Successfully connected to broker"); } - Ok(false) => {} + Ok(_) => {} } match mqtt_eventloop.yield_event(&mut network) { @@ -64,14 +68,14 @@ fn main() { thread::sleep(std::time::Duration::from_millis(5000)); mqtt_client .subscribe(&[SubscribeTopic { - topic_path: format!("{}/device/advisor", thing_name).as_str(), + topic_path: format!("plc/output/{}", thing_name).as_str(), qos: QoS::AtLeastOnce, }]) .unwrap(); mqtt_client .publish( - format!("{}/device/advisor/hello", thing_name).as_str(), + format!("plc/input/{}", thing_name).as_str(), format!("Hello from {}", thing_name).as_bytes(), QoS::AtLeastOnce, ) diff --git a/mqttrust_core/examples/secrets/identity.pfx b/mqttrust_core/examples/secrets/identity.pfx index 625ee6f7f7769c563cf0ce76d2fce76306604aec..254c582ffdec84f1c31f8eeb7efd79f4f55c6565 100644 GIT binary patch literal 2627 zcmai$XEYlM8^?u2Y*k{kRP1XfE?Pv5Qk&W%M%5@f>@7t~T8(JTYeem~6ltlbJ!?y~ zYF@i_QKMSK3c2oi-_!f?et6Dvp5On&|J(08Ff@x6J%9m*X5M3DmcbZe_SpeoKp~p> zFCdzE=?u@o&_LsVBgR5BP~!|;rU%fSE!e*a0Ot1V0I|cYVDNuWHkdf@9I59KA)mO{ zT)?!L$A%lPqNit~gRwC({jZgt8AJyYVr0IFF$8!q00A<*bP~R#C83xTB}Loxvk@H3TDhgyFcK0;zczuKbOVNXvo!eSZ6oNKc(lQX(#98Y6^;1#ruf_5QDBZuKL9s?7Md>TUf%ix)Y2zzueJZ^tjX;v<26HnWpiu) zksAvfqgi=c1X^pL;q5y%;dq~{1Xk%OW8#wPm>fap$eBEx9vcgrii5aY7*49Nn{Rwm zmbMqUgUGcX${halJv^C9b)Mq3t4y!TR~Yp4M52pg-UA!u&#Y zH#7O6pyQ)R&u8PZYJ9e#Qwe#3*9*LFMmi{Sv~dY}E5?vmU$OgYKXn@sJM`5Uw>5)Z zgXd&E#(W{C7%N@ovrHIJWnHkJRT9%45@!6u@>GQ+)R=@ ze`0b8V=!mhYl>h`I*9*mOivAQ^HS$ftg-R<)%d=NZZ{;;z^Ck!yl`jt!o?Qu83m<} zuAZ!!#-C@zqbw~QvUm~a=JuTVBl=E|+C1k*5+J(uWr|(k9HhN=$Cc3YL9&#hjL!x} zrCtFQgK|$q2CXoJzs7@Cm}KMQ@P#cK@5N$QNr!d%>*f}=V6?so=gz|#!~8DJh7$(^ z_xEF_pGeM%K}ylr@`HOSdY#I(j*9OKTt#$ADfZn)*r;>+xc#W5M9lOeFmNu*^NnHdT4TD9}fk4V}3u~TB zOMDJOOzr)e7I|szp5*rtsN%$UXUi347G{XJ9+e-Oi3|#`onZqziED%V$syU6d2t3W zcb2sq-_7i^NBQ)7d)o~6D7EB&aA9<}buojXLCL?o@Q-LfC?FaXd4@yJCXkW!{|;hf z0nnd?YG+X8zr?c9$Jud*tHQ*DKgl>R@sZ$vilx)|_{EBXPG_&I_p<7j{Zk+sIMRhp z|Cx-`=M4TpeOppUZM)#20?EE76VhY$No)6EqW`CfY7LaYN>6F^pR#rS^)tc3u6|`P z7f_%*sUB#%fLk&odptIq{BeNw@iu zLS$wx_Etn~0X^;fu9*Kmgwo7$0i1MU@^IaIdKoh!^yZ>pxir=Z`9@Llqq>rLut=^e z)xJ4>ZoOY|zU~D3TUTySM1p%~LW~v0BMtIUOHL_M;{Z#O|W2 zn2+*`Dd6MM&L;l|-~vUSuoz~{8^!bXh037P7eln2g^LllJci5C%jTH+-A_P)3p%;e zB8yX|mS%1%eJ^;fv5hLOq`wy{x=O6vwei?d(ij|l=>uVHxCrsHKCb-TW2i3G|FPMs zYI;{A^7gD_;~4LDIC4$Af?@;_(yu@4;H+~2KX5WY>Ys~&@40HFL!n%OwQc9Pdj>Gb zP^XJ^wT^G~a!7NPgvLjk#Fg);GSgFlfb1I!EFJcsX4#)_A#3*=4exym&vi3)caen; zVx6_o);DRq_nZANto3tbF;7l^5v$7zXPk&g4Qw$x>yUjlR75Ri3TO6G-ES@5Up0Sn zm&ImB6Ekh`k;C2WC@iGKpIu$D?mhQxX5<6szRaYKXeQSVK=V^s237?B+I-5NEttft zO-Gfx`V67uSxGIxx9ce$lxN8uVnMJu`ZGd#=9TX^OhF{R}dg265j3dm^|^wC?=ZOovnuUVWEL z&0PFYrQR7Nbkp#1+x1S(c9k!EgcuWzqUhESqw>}krVGRyA;8`lCO6lFCcv^61(=f{ z!Qt!yRi6nA%r?QvW^?s%2<*~;WdX6R#L>~hMtK_4On2~Lm^zE?_MOyWDrRjVrY*NF z@LJ$ZtO=|S22Z-DHie0>G=!!5ZO2)Y%Lf4sqw1U-52q)~`HWoEW|I# zdsEDT6qF`eRYpk~^VVf9fACzEp&-crp~{aG7u)@mlN!lq(%B^kPI5c1hZdaDR#Wkk z?+~!>(;rbLl;@iYsP%%{V5isMX7go*Yq?wZBL+e7KT@^6zJLqZGr;0N1hny2(5>J-XaL=hvV!ST7OVNjWyly{0t|NklBECK@ zG!t$5i7hZ8+?&uO0~-7|1KIX#ZM60iz~QLVs#o~s1zwqCr-Y=%?>}i%l7E#Wr07^_ z6JkZ7+0*X6lS!gRy~_f3+u0H$p|r@=$R@N#!StMA%Nxq0hN zeKp|?S5nwj5KIRKhq3>9yXXNRIv})()w}9Qao)VrJ}@0HN2U2hF*b$jc~UMI3a_C& U|4;=o&4nk1`D`h1{iE8y0bxzWN&o-= literal 3349 zcmV+w4eIhRf(;P@0Ru3C4A%w;Duzgg_YDCD0ic2m$OM86#4v&kz%YUbcLoV6hDe6@ z4FLxRpn?ZzFoFkU0s#Opf(Km&2`Yw2hW8Bt2LUh~1_~;MNQU4YUy9RRU>Eak1KnbJF zbZH3pmQYMG>-;1;EP;<=(TsgH^E&GZhwImCT7#>G6e2=6XAy+TQe z0n8*}EAA~~p{nDfOY9(3U@pV-1@ocKg`W{ve%gya;$o?4>0&7m0W3>N#&l{Msi~|% zJG!UsD2NbmkNv};=;4^V1V`tpi{03mL8965XENp-NPeU7KpiyG)&z@m`9M@2G1QIjrEEu`OMM zqp}0&3g;M$H|z&{6@ZXL?|6oqtJ0Ks-0ibF?@Mds3NGS-Ir6uj+vCZ{=26^0q&4?D z`N>+^|*AoO#{tZ!WHdKs~+Ml5uy41$H*HoU)3lF6SkPdIXfn zCw1#j_hq+=uZ=9Gp{NR?4G6mgF$x+wFEnj$BloDd8!#PZH9d8K*N3Jf0sv(>>mIsUnopONQF>S``9}$yi+oN?BsZjQBM~XC9QnQ{(!I5t<#auK*$}vLO?BrZgR_ zckfk`tDfux*wWq<|#ZVi>}Pjz2&{Rad6rbd)(rj2wLqY8|*ifel3%wnBbX-Jc!T z1?BQa#5zJ04q@z?e8tF{iLVn73~Mz8CY*y)1=iV8IX#6w<36`aD3-f^uIV>B<0GB> z_QnfF4{o1}xqV!5ig<#ed!;d(oJ`kuQPA<(TyCPgy|1?Zrb-B#Z9ARzK+!K9!7j|&@fKIldNiV#Ua2_sIx2Wz0kr^ZAHuYcY zrc<u?_2b7>LW1}UE7#xa!7-}IYxlVI2F%qYq*0h5cPj*Kw_jNc;v!1P zd^AB#!C?yV>E)uukVk6+8+f1qm*1ffRPKM`hh|sUF`$?9IgWcZvS<=be8ib1NbqUb zYx*2t;D-L((0D*SH$$q5970I@3#Jo;nHywCw1I}?Y^K1?PB$0^d@L?+4;$0>(J*&Q z0Q3#5(V2Psl+!E!{!7U8sB7%bISt1`Zh9hl{1pw`8Izh{mDiXY0edXF3$D#{xR1FYkq+>BdLHBj&*uQ^G{(^EiwA#pyQCuuAek!ZJQ|X}U3? ze0`9IRQocAB5xb$a`jieSyrzZzNdZSU5A#N;EkTZ9dif`nexl~e)FUoBx9tSg&N$a zRgN@U_bTdey0fMjzZb@=a=p(_OeVF6&QnR$f3SGdTG)0A19QKR90qL^gjb!b2fo0| z^l~>AU^Wt8e@Nl^1hVFAGd9^KNqZc9qyD*(QbgOmwJ`>K(XFZhHwVKG;gwrlevWUZ zHj8hWBerZdQYPCPPDc~O@?Yp8{Ahvi)9XDRO#plwf{aw9&|n+5B6W%nQ93RLePqxm z&@jNjNnjfandX;}DkDa4*HTVzBkL`dq?H2`Cwk)~qWfTI?Q$Yxb?x0DiT41LFi0Sk zyuFp6JR06}waAwW!_FPP`!ymv$h_!sb9-aPiM`TsepcR8%y}^mL}hEPh}a;)LV2mh zVWqKQk2S>(=;M`T&Z&xW;02}Tnq<$xtVlfo#Z|vI`xJ?QAJ+#ahIn0~8p9;YdY`vT ztsZKV>^A*8EAo%v9LiT&G3tT2#pQ~qxf*y#5DjkXhQuIA`urVF@!YuhB=Y?aR-{Ht zaz@{8#%)o!vc)p?M2p1L@)LL`WFi+#9&v^ba|j&$+M$}9rYbbn59fhvZGK@i*VO$H z;Ee*(_vE7}147a!tzJA!@#lBl8&s`4T!D)CgqA-VFoFd^1_>&LNQU~S-=7U2ml0v1jq+F$;uGIzMl^p8&^l=uvyIaU|+fxa7+hW#I88Xbmak9qQBg8 z_s&R&+Q`5vaImK?F@_3^6ovopwuVqbj6{vMgKjjG3M9-@YoEGT5c819!vJ9v6tds0 zTx`2(wg+AO-ndsI-0Wp0T(p($S0w|0q7q9F#mixb3lf=vQx&+EJ8itjX9mF*oDPiO zZniL8H|%U8rBbXr{}<&9FIX(diu)jHtG@D&N$)=FdA_gpAHGE+sjdOTmky5`FSiys zm%@lk>bUI zxxL~6V)VN@P2vqb?e$e#XLZIqhZat2+m2lI-0b~+pne>q@C9h(d0(@R&elX zZ3cc7T;$$j5-1MO8ZpOGDLrmy+^eG;ZSXkygvf?l_O0u4OEX7g_8M{FI3A&GM0op~ z*C5+7cakUbfXDmL*{&!*XrQ&o{)*vXp(23c#rM>N=unCQ-|$~>96izlwQk>ceTv-P zP}|%C#zSb1O=fu;1UsW1S@ayh~@ zi?Q+HF0ZU|a5aR%I0oN#u80_Z^2d?93p^-r)_S|qlxzu#tk!80JRi*N|oU&9ExK={|TN;q^XVq z>8w6V81F$H{jOG&cP)-B;xU?!gD;D$VY!JqLFNyVApU2&bT%!YFuc6@a{vs@ zB;dA?EHe{gYDLDR_#XpLWl~f{ak1Xs+aG)WfD;|-c5AuaXSg+WijuuVL}lrJqEQE(nW0@@3fsVWi>9_f7Vg(^v90?pOiC;UlfCAy13Sb z1KTMtcOOl_(MJlwg5w!{Y>;;LcU-40&t3O2)2uyg+&>X*QqO`af($5B)t=CG>0rP2 z8y0vT{vy)1&2*07>U1da&HG70aGrankweV!sFQcr*30|w!@zz5evV(%XwTHnuQc;y zn>FxJflNvJC4-Vgr{>xY=O26ih)o@dqPd;w*~^WMWU!)Y@HNUV8a;XDx;r(IJplTa zw73q(M}jZM#P7#d%Fu#H4fJA~3`v35b(Vqrxq^g!Z83rG_1x_|ElMyNG}1R>eqZml zuDxf1tRhU_lqt?t3x7wFpPdvrKYG>bq)X#6-E*F{;1@&>8GKUXkTEeOFe3&DDuzgg z_YDCF6)_eB6o^6oG4Lj(#+PE|syJ#A6Wch>, network_handle: NetworkHandle, + connect_counter: u8, } impl<'a, 'b, S, O, const TIMER_HZ: u32, const L: usize> EventLoop<'a, 'b, S, O, TIMER_HZ, L> @@ -41,6 +42,7 @@ where options, requests: Some(requests), network_handle: NetworkHandle::new(), + connect_counter: 0, } } @@ -54,7 +56,7 @@ where pub fn connect + ?Sized>( &mut self, network: &mut N, - ) -> nb::Result { + ) -> nb::Result, EventError> { // connect to the broker match self.network_handle.is_connected(network) { Ok(false) => { @@ -216,16 +218,23 @@ where } } + fn backoff(&self) -> TimerDurationU32 { + let base_time_ms: u32 = 1000; + let backoff = base_time_ms.saturating_mul(u32::pow(2, self.connect_counter as u32 - 1)); + + core::cmp::min(50.secs(), backoff.millis()) + } + fn mqtt_connect + ?Sized>( &mut self, network: &mut N, - ) -> nb::Result { + ) -> nb::Result, EventError> { match self.state.connection_status { - MqttConnectionStatus::Connected => Ok(false), + MqttConnectionStatus::Connected => Ok(None), MqttConnectionStatus::Disconnected => { - info!("MQTT connecting.."); let now = self.last_outgoing_timer.now(); self.state.last_ping_entry().insert(now); + self.connect_counter += 1; self.state.await_pingresp = false; self.network_handle.rx_buf.init(); @@ -242,6 +251,12 @@ where password, }); + info!( + "MQTT connecting.. Attempt: {}. Backoff time: {}", + self.connect_counter, + self.backoff().to_millis() + ); + // mqtt connection with timeout self.network_handle.send_packet(network, &connect)?; self.state.handle_outgoing_connect(); @@ -250,16 +265,19 @@ where MqttConnectionStatus::Handshake => { let now = self.last_outgoing_timer.now(); + let backoff_time = core::cmp::max(50.secs(), self.backoff()); + if self .state .last_ping_entry() .or_insert(now) - .has_elapsed(&now, 50.secs()) + .has_elapsed(&now, backoff_time) { return Err(nb::Error::Other(EventError::Timeout)); } - self.network_handle + let res = self + .network_handle .receive(network) .map_err(|e| e.map(EventError::Network))? .decode(&mut self.state) @@ -267,8 +285,16 @@ where if n.is_none() && p.is_none() { return Err(nb::Error::WouldBlock); } - Ok(n.map(|n| n == Notification::ConnAck).unwrap_or(false)) - }) + Ok(n) + }); + + match res { + Ok(r) => { + self.connect_counter = 0; + Ok(r) + } + Err(e) => Err(e), + } } } } @@ -417,7 +443,7 @@ impl NetworkHandle { #[derive(Debug)] struct PacketBuffer { range: RangeTo, - buffer: Vec, + buffer: Vec, } impl PacketBuffer { diff --git a/mqttrust_core/src/lib.rs b/mqttrust_core/src/lib.rs index df72d34..285bb57 100644 --- a/mqttrust_core/src/lib.rs +++ b/mqttrust_core/src/lib.rs @@ -18,6 +18,7 @@ pub use client::Client; use core::convert::TryFrom; pub use eventloop::EventLoop; use heapless::{String, Vec}; +use mqttrust::encoding::v4::Connack; pub use mqttrust::encoding::v4::{Pid, Publish, QoS, QosPid, Suback}; pub use mqttrust::*; pub use options::{Broker, MqttOptions}; @@ -39,7 +40,7 @@ pub struct PublishNotification { // #[cfg_attr(feature = "defmt-impl", derive(defmt::Format))] pub enum Notification { /// Incoming connection acknowledge - ConnAck, + ConnAck(Connack), /// Incoming publish from the broker #[cfg(not(feature = "std"))] Publish(heapless::pool::singleton::Box), diff --git a/mqttrust_core/src/state.rs b/mqttrust_core/src/state.rs index f4d1087..441c472 100644 --- a/mqttrust_core/src/state.rs +++ b/mqttrust_core/src/state.rs @@ -145,7 +145,7 @@ impl MqttState { match packet { Packet::Connack(connack) => self .handle_incoming_connack(connack) - .map(|()| (Notification::ConnAck.into(), None)), + .map(|()| (Notification::ConnAck(connack).into(), None)), Packet::Pingresp => self.handle_incoming_pingresp(), Packet::Publish(publish) => self.handle_incoming_publish(publish), Packet::Suback(suback) => self.handle_incoming_suback(suback), @@ -269,12 +269,23 @@ impl MqttState { let qospid = (publish.qos, publish.pid); #[cfg(not(feature = "std"))] - let boxed_publish = BoxedPublish::alloc().unwrap(); - #[cfg(not(feature = "std"))] - let notification = Notification::Publish(boxed_publish.init(publish.try_into().unwrap())); + let notification = if publish.payload.len() > 4096 { + error!( + "Received payload larger the {}! Sending ACK but discarding payload notification!", + 4096 + ); + None + } else { + let boxed_publish = BoxedPublish::alloc().unwrap(); + Some(Notification::Publish( + boxed_publish.init(publish.try_into().unwrap()), + )) + }; #[cfg(feature = "std")] - let notification = Notification::Publish(std::boxed::Box::new(publish.try_into().unwrap())); + let notification = Some(Notification::Publish(std::boxed::Box::new( + publish.try_into().unwrap(), + ))); let request = match qospid { (QoS::AtMostOnce, _) => None, @@ -289,7 +300,7 @@ impl MqttState { } _ => return Err(StateError::InvalidHeader), }; - Ok((Some(notification), request)) + Ok((notification, request)) } fn handle_incoming_pubrel( diff --git a/.github/scripts/da_monitor.sh b/scripts/da_monitor.sh similarity index 100% rename from .github/scripts/da_monitor.sh rename to scripts/da_monitor.sh diff --git a/scripts/rotate_secrets.sh b/scripts/rotate_secrets.sh new file mode 100755 index 0000000..d99bf21 --- /dev/null +++ b/scripts/rotate_secrets.sh @@ -0,0 +1,27 @@ +#!/usr/bin/env bash + +if [[ -z "$DEVICE_ADVISOR_PASSWORD" ]]; then + echo "DEVICE_ADVISOR_PASSWORD environment variable must be set!" + return 1; +fi + +SCRIPT_DIR="$(dirname "$(readlink -f "$0")")" +SECRETS_DIR="$SCRIPT_DIR/../mqttrust_core/examples/secrets" +THING_NAME="mqttrust" + +CERT_PATH="$SECRETS_DIR/cert.pem" +PRIV_KEY_PATH="$SECRETS_DIR/priv.key.pem" + +CERT_ARN=$(aws iot create-keys-and-certificate --set-as-active --certificate-pem-outfile $CERT_PATH --private-key-outfile $PRIV_KEY_PATH | jq -r .certificateArn); +aws iot attach-thing-principal --thing-name $THING_NAME --principal $CERT_ARN > /dev/null 2>&1 +aws iot attach-policy --policy-name Connect --target $CERT_ARN > /dev/null 2>&1 +aws iot attach-policy --policy-name Input --target $CERT_ARN > /dev/null 2>&1 +aws iot attach-policy --policy-name Output --target $CERT_ARN > /dev/null 2>&1 + +rm $SECRETS_DIR/identity.pfx + +# Generate new identity.pfx +openssl pkcs12 -export -passout pass:"$DEVICE_ADVISOR_PASSWORD" -out $SECRETS_DIR/identity.pfx -inkey $PRIV_KEY_PATH -in $CERT_PATH + +rm $CERT_PATH +rm $PRIV_KEY_PATH \ No newline at end of file From bb3e567a9ef335661c7407431135a42b77d7c754 Mon Sep 17 00:00:00 2001 From: Mathias Date: Thu, 2 May 2024 19:14:07 +0200 Subject: [PATCH 2/3] Fix tests --- mqttrust_core/src/eventloop.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/mqttrust_core/src/eventloop.rs b/mqttrust_core/src/eventloop.rs index 64e5500..42b246d 100644 --- a/mqttrust_core/src/eventloop.rs +++ b/mqttrust_core/src/eventloop.rs @@ -220,7 +220,7 @@ where fn backoff(&self) -> TimerDurationU32 { let base_time_ms: u32 = 1000; - let backoff = base_time_ms.saturating_mul(u32::pow(2, self.connect_counter as u32 - 1)); + let backoff = base_time_ms.saturating_mul(u32::pow(2, self.connect_counter as u32)); core::cmp::min(50.secs(), backoff.millis()) } @@ -725,12 +725,12 @@ mod tests { rx_buf.range.end += connack_len; let publish_len = encode_slice(&Packet::from(publish.clone()), rx_buf.buffer()).unwrap(); rx_buf.range.end += publish_len; - assert_eq!(rx_buf.range.end, rx_buf.buffer.capacity()); + assert_eq!(rx_buf.range.end, 4096); // Decode the first Connack packet on the Handshake state. state.connection_status = MqttConnectionStatus::Handshake; let (n, p) = PacketDecoder::new(&mut rx_buf).decode(&mut state).unwrap(); - assert_eq!(n, Some(Notification::ConnAck)); + assert!(matches!(n, Some(Notification::ConnAck(_)))); assert_eq!(p, None); let mut pkg = SerializedPacket(&mut rx_buf.buffer[rx_buf.range]); @@ -776,7 +776,7 @@ mod tests { rx_buf.range.end += connack_malformed_len; let publish_len = encode_slice(&Packet::from(publish.clone()), rx_buf.buffer()).unwrap(); rx_buf.range.end += publish_len; - assert_eq!(rx_buf.range.end, rx_buf.buffer.capacity()); + assert_eq!(rx_buf.range.end, 4096); // When a packet is malformed, we cannot tell its length. The decoder // discards the entire buffer. From 59b0717d86274b755e836ca6e67954e13097175c Mon Sep 17 00:00:00 2001 From: Mathias Date: Fri, 3 May 2024 12:09:34 +0200 Subject: [PATCH 3/3] Pass all qualification tests in CI --- .gitignore | 1 + mqttrust_core/Cargo.toml | 7 ++- mqttrust_core/examples/aws_device_advisor.rs | 61 ++++++++++++------- mqttrust_core/examples/common/clock.rs | 2 - mqttrust_core/examples/common/credentials.rs | 4 +- mqttrust_core/examples/secrets/identity.pfx | Bin 2627 -> 2627 bytes mqttrust_core/src/eventloop.rs | 6 +- scripts/da_monitor.sh | 14 ++++- scripts/da_run_test.sh | 49 +++++++++++++++ scripts/rotate_secrets.sh | 6 ++ 10 files changed, 121 insertions(+), 29 deletions(-) mode change 100644 => 100755 scripts/da_monitor.sh create mode 100755 scripts/da_run_test.sh diff --git a/.gitignore b/.gitignore index 59a4524..0d3b7c6 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,4 @@ .gdb_history Cargo.lock target/ +device_advisor_integration.log \ No newline at end of file diff --git a/mqttrust_core/Cargo.toml b/mqttrust_core/Cargo.toml index 16cade3..f6c54af 100644 --- a/mqttrust_core/Cargo.toml +++ b/mqttrust_core/Cargo.toml @@ -40,11 +40,14 @@ defmt = { version = "^0.3", optional = true } [dev-dependencies] native-tls = { version = "^0.2" } dns-lookup = "1.0.3" -env_logger = "0.9.0" +env_logger = "0.11" +static_cell = "2.1" [features] default = [] std = [] -defmt-impl = ["defmt", "mqttrust/defmt-impl", "heapless/defmt-impl", "fugit/defmt"] +log = ["dep:log", "mqttrust/log"] + +defmt-impl = ["dep:defmt", "mqttrust/defmt-impl", "heapless/defmt-impl", "fugit/defmt"] diff --git a/mqttrust_core/examples/aws_device_advisor.rs b/mqttrust_core/examples/aws_device_advisor.rs index 8720095..9840706 100644 --- a/mqttrust_core/examples/aws_device_advisor.rs +++ b/mqttrust_core/examples/aws_device_advisor.rs @@ -8,19 +8,27 @@ use mqttrust_core::{EventLoop, MqttOptions, Notification}; use common::clock::SysClock; use common::network::Network; use native_tls::TlsConnector; +use static_cell::StaticCell; +use std::sync::atomic::AtomicBool; use std::sync::Arc; use std::thread; use crate::common::credentials; -static mut Q: BBBuffer<{ 1024 * 6 }> = BBBuffer::new(); +static mut Q: BBBuffer<{ 1024 * 60 }> = BBBuffer::new(); fn main() { env_logger::init(); let (p, c) = unsafe { Q.try_split_framed().unwrap() }; - let hostname = credentials::HOSTNAME.unwrap(); + static HOSTNAME: StaticCell = StaticCell::new(); + let hostname = HOSTNAME.init(credentials::hostname()); + + log::info!( + "Starting device advisor test on endpoint {}", + hostname.as_str() + ); let connector = TlsConnector::builder() .identity(credentials::identity()) @@ -28,18 +36,21 @@ fn main() { .build() .unwrap(); - let mut network = Network::new_tls(connector, String::from(hostname)); + let mut network = Network::new_tls(connector, hostname.clone()); let thing_name = "mqttrust"; let mut mqtt_eventloop = EventLoop::new( c, SysClock::new(), - MqttOptions::new(thing_name, hostname.into(), 8883), + MqttOptions::new(thing_name, hostname.as_str().into(), 8883), ); let mqtt_client = mqttrust_core::Client::new(p, thing_name); + let connected = Arc::new(AtomicBool::new(false)); + let con = connected.clone(); + thread::Builder::new() .name("eventloop".to_string()) .spawn(move || loop { @@ -49,12 +60,18 @@ fn main() { session_present, code: ConnectReturnCode::Accepted, }))) => { - log::info!("Successfully connected to broker"); + log::info!( + "Successfully connected to broker. session_present: {}", + session_present + ); + con.store(true, std::sync::atomic::Ordering::Release); + } + Ok(n) => { + log::info!("Received {:?} during connect", n); } - Ok(_) => {} } - match mqtt_eventloop.yield_event(&mut network) { + match nb::block!(mqtt_eventloop.yield_event(&mut network)) { Ok(Notification::Publish(_)) => {} Ok(n) => { log::trace!("{:?}", n); @@ -66,19 +83,21 @@ fn main() { loop { thread::sleep(std::time::Duration::from_millis(5000)); - mqtt_client - .subscribe(&[SubscribeTopic { - topic_path: format!("plc/output/{}", thing_name).as_str(), - qos: QoS::AtLeastOnce, - }]) - .unwrap(); - - mqtt_client - .publish( - format!("plc/input/{}", thing_name).as_str(), - format!("Hello from {}", thing_name).as_bytes(), - QoS::AtLeastOnce, - ) - .unwrap(); + if connected.load(std::sync::atomic::Ordering::Acquire) { + mqtt_client + .subscribe(&[SubscribeTopic { + topic_path: format!("plc/output/{}", thing_name).as_str(), + qos: QoS::AtLeastOnce, + }]) + .unwrap(); + + mqtt_client + .publish( + format!("plc/input/{}", thing_name).as_str(), + format!("Hello from {}", thing_name).as_bytes(), + QoS::AtLeastOnce, + ) + .unwrap(); + } } } diff --git a/mqttrust_core/examples/common/clock.rs b/mqttrust_core/examples/common/clock.rs index 227f6e6..2d19c5e 100644 --- a/mqttrust_core/examples/common/clock.rs +++ b/mqttrust_core/examples/common/clock.rs @@ -5,14 +5,12 @@ use std::{ }; pub struct SysClock { start_time: u32, - countdown_end: Option, } impl SysClock { pub fn new() -> Self { Self { start_time: Self::epoch(), - countdown_end: None, } } diff --git a/mqttrust_core/examples/common/credentials.rs b/mqttrust_core/examples/common/credentials.rs index 3057593..22bacb9 100644 --- a/mqttrust_core/examples/common/credentials.rs +++ b/mqttrust_core/examples/common/credentials.rs @@ -11,4 +11,6 @@ pub fn root_ca() -> Certificate { Certificate::from_pem(include_bytes!("../secrets/root-ca.pem")).unwrap() } -pub const HOSTNAME: Option<&'static str> = option_env!("AWS_HOSTNAME"); +pub fn hostname() -> String { + env::var("AWS_HOSTNAME").unwrap() +} diff --git a/mqttrust_core/examples/secrets/identity.pfx b/mqttrust_core/examples/secrets/identity.pfx index 254c582ffdec84f1c31f8eeb7efd79f4f55c6565..3fb464ec76d9964df117ce121d0b1425a8d92eac 100644 GIT binary patch delta 2390 zcmV-c390tO6vGsdYY>Xp;kgs4sq^oZIIJ(Pit&+qA%CdbCa|8jaR%RmS&Kg8v@d{y z1HiEgGW6V%q!86{k=6=+MjUmjxy<}S^^DrS+IaKTK&9;FTbrX_7^&&vQ?_wqo-b2n zXEU;1ngLZ_I%bP7j*X%(LP0|M*1(Fe3GDJtEp(|R9ZLWY&zCp(Poxrj=vD-l+b)7S z5qm9OJAaWfnk8@1fa>fL1AxjC_J)~q;})&Y2A^xic!W6~VRcBW^sxPpZWQLYJ9q8yJi3IC{qS-`ew*7G!PGp!D|LLr z+IPrmTlHM|mdfiFV|2Z6WZkf*EpXcM#HNN|dd2maT9YPsm~+?s9$y98}sj zS9kmxkVW9nQHk)_DJ(4*O(Z(+m; z0IX2`L{3Nt@#47BpPl$Rh2s`RUM}I_J^8$)uaKszV$_IoC;Dj&x9loaBwrbCYEO0c z0}ZJl)|%g0T;@qfA>S8@CwJTrhJRbXVwjUFI34PZ$|M}nRUWVCwG20-;0?gXV?>k#A}xIXP`YtE7B6x9^E+f%I#RK zE6d{+Qgj}ya9|B&Xc$f#iRs7GD9rH3gu<-EN6YDLD;~$5-)UpP_r8|5C0jsR%{yaA zPVI$^D12VS8*%Sq%Mf>rm{C)DefcM4Uowl2+?sWhVCuAyLD}nAyt1vq(~ewyvJpL2 zg36z0vc?`4``w4xUbSZ;O&}b!IS3_vSeUT`uotdT-ji?yR1nc2cY24u5Mj7qTi`z+ zIE0aVA%9AOdh+K{rk(DozV9;#;X(w01kikCf5`um`3GgG=t@bGfl%DYw-a0uT`!Szn%}w)t|{iO--fFS)KmD3{rn` z3@JCW-G0?Z=%s1GgkC9kgdz?kO*dvD^2-1PB7g3uHNC^Vq-c8~8Rz+A6_Z8Bvt}yi zzp2jWDI+M~`btU6>XcNPKk{Rt6QnN7C({6GEm)AbGfRVhvrm z>}55ZFyd~~6-|nP1VOiiC^mb#CSAySxqnZdZ&LE^dA>{fY4J8%i5!Va=h`HM7ODxJ zab&bc8=a~F+aGEikiWW_Nda)hoY{3v!DGdOv^PfWG1ZnuLn6HDg<4kemy^I2d5e2gc@nPokzmaSlvLYcFl2V`pDacF{V@yy`agsVjDR~sS;D(n zRd`_AyJ)hI6eg7kgo<8cl!&OZK3o2_EL=%6XnGgd`5iO)$;L*df%*PDbQ=CoxyVgx z%39uUdL@bX0-|=2VmT+i8GlXq^zvk7``QO++gAye#8vK9Fe}4_^rBpKI%6@AL zk2Ep71t6=4x3_-C;B8V1L%~%}txqIs)iIVDHJ%%u9m`NvYnbgz>&87nePZ06r|(Nu zxIa(2HHLLzXcd$}A=7iT-qbiSGkY>ug9ksj*MOWV(L03{o*akp9YX`6Fg|q>?IzaO zZ#=9Up&%*aLY^&yTz`#buCu76XjHAI_pMm}D8ha{Py_hmnO`FShK)R^uR(|&yf%

iN(SHpP>z{{1r4&kB@ADu#UG3 z2FuymYCk=RmcU;FG85nYO`?G@B`_lf2`Yw2hW8Bt2^BFG1Qir_4X|;DD=qzFzWolf z9K~h+`1rVz3KcF4%lUVLZOaHBLH4j>+PLp`)*dXuA6e`j{M|eKx|-bt2nRjHFsIpl I*8&0v02kwrtN;K2 delta 2390 zcmV-c390tO6vGsdYY?`^hn$_S^}L@5;hPMN)~%6yA%CX^DkVxB2Kq(Vm_2=NM#g}G z1Hhj!1Y)Wm4ru7kBhEV4>AQ}{^=$5)D;69byk-l4(V`DiSRmxH!3uK>#1|gfiV-O} zle32NWRrcnjZjnn_@XWy0wo}5B6g#055{*+l-BTJ(>EA>;Xtcn&gB4G=!~K8{G?Eg zy59`gE`LD?>GsKutO~P%izKz)?ARiA{!;AO?v#j?R((5tk&Lle0>foJc;n1KiP(>% zR{NUkEzFSO^qz)$TvTM8DvpG=U-zzZE0#!FA09s?FPVBL|Ltm29LtvHVnn&nbsV-Q z;R@}U>0-H1tzmIJ!+J4_W2+>$co_NlDaAT`Z+08NW6217`1G)_}* z&`{tG)0Ue*od3RYNM~zQcaUh7gbOLzO0VKJ)hwJr`drSBGVe z69apYKUkV+kj(wsXv*A2YcP;8NkSN%1Shfqt2NyNjqI)7Yov7XoL&u*!k|0?61%5P z8m`0%&Vw9Thk0b4)+;#|Rdv~fn;b@;W`AILSTqj55glePe6|a#4QW7+U(h4@X*1>GL3f-y%{_wr56U7SJqMY? z1h^-dnU0KGOIwlr2%!iSv(pq+Hal)iKQtP|qttMQ!oE826pwlFh|nTEbhn-*>3?^? zbG)`D68%dahDY8^$4op8iTHsi)kHZwU>?HEe@aT$K7q*B#9vzo%F>wqn|EAcpy9m} z+%7WFUStcEO4QsWdQnkWmKHn^-t}AfQXKrQ`J@#E>(sVsV$4QhF;FATXvq9>2-R*;q2 z`$mwo)t|?&18;r6ra%ONTT?=9Xa%ue2ic6x6c$S27Folovf}JgthU3PL0daV;kNi0 zj{Cg}n_azunLHyo#pyrJ;@)j0@6aGG*Ve+lI0*^~6(mIb7$B8fa&T4L3{J$DqH5Hg`Ds;w8 z(m(Wrg>TZ=t~-4h=f3dUTUR$8hr)z{3$!L$6eePz)EHmV(ZYAi%HZd zBF=@WB{}m!8*LGYi#^wK-w66K50F`3iDwsybbl(`8^@}8?gV63YB)|a%zZtXT><^W zhMSGV^+c51Ap%Y!uvmNfRxG5m9AS`~2@}87>hZ*z>(BjWqnEmGCm}pQbzTeWI)aZGR%@ zudMV`WAidQ%hRZ66$`l=6>3xatTSWGwUKX#P3Sh0!@65%-C4TT7Vvjl=_IQ5NfjMM zxB0;jwOt2(T1Q(&5PUK9UO$r{AQ5r1zkd)D#>$3Uby^#>vstu6nYP}noVthYujlW5 zrA+<-87r|*ivWxGvgL0V;Fk32|Bz;4kP%>i^x+B@6;qb%zsT51^(a@(Fuv4zq7<3Q zp$lE-4}c*dvM}M8miZ0nbs;%?*^N!8I%?^iZ#JFS^a@n-J%-v((GFov`gU}^Z+{Is zCbh*A-IaNMT*#G)!Fvc^!2!IFD+3EW80p`R+r5T`mhD+JY()cg&fO+I0qRDMWA>hm zT^^t;^LTnGIn)Wn{~wg7cw}*Nk^h!*JVbj|>wEO#ke+?JklAx?G&lhYe(ULDx5nH= zMD<>2W|A*Lv&jQ0hPfnZ%2%%kzkedN66BRwkp+-C^v?X1;m@r)!g3v8NH@Pm!acw> z<{Tw@?Q+;MT3ON}d-5y%c1 z^#25% zFp@k`VP{$0OE>s%xzjoi4%SiZW!R)I?D;0Z%!At4seoE?R|Rtwy!Q9`WXZn{9Wh9S z%^{00fAYZ5Pg_*V(wc^~wgDQt^?fe!YeSA5OQP4HvS4^16{7yGCv%;IjVeRD<4l}O zqsZmOwOi<$Ixz5oQ?IT#Wa*`&krS^%{ zWD>n4|bwEc39;;98RCk&q_1JTULt(S1wyrtL0$ z>z}i8TCfMaPUS8`n(l6R%5{~z`L95^asWIc>u*RC7-NZ#|C)aHdw*KJM;x-*Qv+}W zEUY3c_8}PXuN!2n-TtS{1trxwYLqQq%9J4mY3tu=q|WF2e?87Lm4Jcre^-`Hn-MU! z21osDWkT1Ww(6Mi+C8nwUEvkoDvVgI1y+LkU1@<6I1_C0JTaFqqyL}6-uwr)w)8az z7Puh(!_umyB!dsvc7G*`<&ur2{EB>K9$_Z}cVn=?_*?5da~*i(h@ed(yYh>lOPp=j zJ>$rVQJWj zB1y;P8En7|g?S+VdFXk&f*pl1B`_lf2`Yw2hW8Bt2^BFG1Qis&wK-+TL#Qm@Q|(1L zAM9SXc0L7?3KcH93udkQFW*kAS8 TimerDurationU32 { - let base_time_ms: u32 = 1000; + let base_time_ms: u32 = 200; let backoff = base_time_ms.saturating_mul(u32::pow(2, self.connect_counter as u32)); core::cmp::min(50.secs(), backoff.millis()) @@ -265,7 +265,7 @@ where MqttConnectionStatus::Handshake => { let now = self.last_outgoing_timer.now(); - let backoff_time = core::cmp::max(50.secs(), self.backoff()); + let backoff_time = self.backoff(); if self .state @@ -273,6 +273,7 @@ where .or_insert(now) .has_elapsed(&now, backoff_time) { + warn!("Timed out waiting for connect packet!"); return Err(nb::Error::Other(EventError::Timeout)); } @@ -562,6 +563,7 @@ impl<'a> PacketDecoder<'a> { Err(EventError::Encoding(e).into()) } Ok(Some(packet)) => { + warn!("Got packet! {:?}", packet); self.is_err.replace(false); state .handle_incoming_packet(packet) diff --git a/scripts/da_monitor.sh b/scripts/da_monitor.sh old mode 100644 new mode 100755 index 508b298..44ab76a --- a/scripts/da_monitor.sh +++ b/scripts/da_monitor.sh @@ -44,6 +44,15 @@ function report_status { done } +function check_pid { + if [ -n "$pid" ]; then + ps -p $pid > /dev/null; + return $? + else + return 0 + fi +} + while test ${IN_PROGRESS} == 1; do # Fetch the current status and stash in /tmp so we can use it throughout the status fetch process. @@ -68,13 +77,16 @@ while test ${IN_PROGRESS} == 1; do elif test x"${overall_status}" == x${STATUS_PASS}; then MONITOR_STATUS=0 IN_PROGRESS=0 + elif test x"${overall_status}" == x${STATUS_PASS_WITH_WARNINGS}; then + MONITOR_STATUS=0 + IN_PROGRESS=0 elif test x"${overall_status}" == x${STATUS_STOPPING}; then MONITOR_STATUS=1 IN_PROGRESS=0 elif test x"${overall_status}" == x${STATUS_STOPPED}; then MONITOR_STATUS=1 IN_PROGRESS=0 - elif { ps -p $pid > /dev/null; }; [ "$?" = 1 ]; then + elif check_pid; [ "$?" = 1 ]; then echo Binary is not running any more? MONITOR_STATUS=1 diff --git a/scripts/da_run_test.sh b/scripts/da_run_test.sh new file mode 100755 index 0000000..8f350ea --- /dev/null +++ b/scripts/da_run_test.sh @@ -0,0 +1,49 @@ +#!/usr/bin/env bash + +if [[ -z "$DEVICE_ADVISOR_PASSWORD" ]]; then + echo "DEVICE_ADVISOR_PASSWORD environment variable must be set!" + return 1; +fi + +set -e + +DAEMONIZE=true + +THING_NAME="mqttrust" +SUITE_ID="1gaev57dq6i5" +export RUST_LOG=debug + +THING_ARN="arn:aws:iot:eu-west-1:411974994697:thing/$THING_NAME" +SCRIPT_DIR="$(dirname "$(readlink -f "$0")")" + + +export AWS_HOSTNAME=$(aws iotdeviceadvisor get-endpoint --thing-arn $THING_ARN --output text --query endpoint) + +cargo build --features=log --example aws_device_advisor --release + +SUITE_RUN_ID=$(aws iotdeviceadvisor start-suite-run --suite-definition-id $SUITE_ID --suite-run-configuration "primaryDevice={thingArn=$THING_ARN},parallelRun=true" --output text --query suiteRunId) +if $DAEMONIZE; then + nohup ./target/release/examples/aws_device_advisor > device_advisor_integration.log & + PID=$! +else + echo "You can now run 'DEVICE_ADVISOR_PASSWORD=$DEVICE_ADVISOR_PASSWORD AWS_HOSTNAME=$AWS_HOSTNAME ./target/release/examples/aws_device_advisor' in a seperate terminal" +fi + +always() { + kill $PID || true + cat device_advisor_integration.log +} + +on_failure() { + if $DAEMONIZE; then + always || true + fi + aws iotdeviceadvisor stop-suite-run --suite-definition-id $SUITE_ID --suite-run-id $SUITE_RUN_ID +} + +trap "on_failure" ERR INT + +$SCRIPT_DIR/da_monitor.sh $SUITE_ID $SUITE_RUN_ID $PID + +always + diff --git a/scripts/rotate_secrets.sh b/scripts/rotate_secrets.sh index d99bf21..5a06044 100755 --- a/scripts/rotate_secrets.sh +++ b/scripts/rotate_secrets.sh @@ -13,6 +13,12 @@ CERT_PATH="$SECRETS_DIR/cert.pem" PRIV_KEY_PATH="$SECRETS_DIR/priv.key.pem" CERT_ARN=$(aws iot create-keys-and-certificate --set-as-active --certificate-pem-outfile $CERT_PATH --private-key-outfile $PRIV_KEY_PATH | jq -r .certificateArn); +for OLD_CERT in $(aws iot list-thing-principals --thing-name $THING_NAME | jq -r '.principals[]' | xargs); do + CERT_ID=$(echo $OLD_CERT | cut -d "/" -f 2) + aws iot detach-thing-principal --thing-name $THING_NAME --principal $OLD_CERT + aws iot update-certificate --new-status INACTIVE --certificate-id $CERT_ID + aws iot delete-certificate --certificate-id $CERT_ID --force-delete +done aws iot attach-thing-principal --thing-name $THING_NAME --principal $CERT_ARN > /dev/null 2>&1 aws iot attach-policy --policy-name Connect --target $CERT_ARN > /dev/null 2>&1 aws iot attach-policy --policy-name Input --target $CERT_ARN > /dev/null 2>&1