Not able to install snaps using side loading rest api command

snapd
1

Hi, I am trying to side load a snap using rest api and this is my script

#!/usr/bin/env python3

# -*- coding: utf-8 -*-
import socket
import json
import requests
from urllib3.connection import HTTPConnection
from urllib3.connectionpool import HTTPConnectionPool
from requests.adapters import HTTPAdapter
import os

class SnapdConnection(HTTPConnection):
    def __init__(self):
        super().__init__("localhost")

    def connect(self):
        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.sock.connect("/run/snapd.socket")


class SnapdConnectionPool(HTTPConnectionPool):
    def __init__(self):
        super().__init__("localhost")

    def _new_conn(self):
        return SnapdConnection()


class SnapdAdapter(HTTPAdapter):
    def get_connection(self, url, proxies=None):
        return SnapdConnectionPool()

class RefreshSnap():

    def __init__(self):
        #SET GLOBAL SESSION AND MOUNT CUSTOM ADAPTER
        self.session = requests.Session()
        self.session.mount("http://snapd/", SnapdAdapter())
        self.intall_snap()

    def intall_snap(self):
        try:
            snap_file_name = "turftank-mk1-sw_2.1.2_arm64.snap"
            with open(os.environ["SNAP_COMMON"] + "/" + snap_file_name,'rb') as f:
                file_data = f.read()
            bytes_string = file_data.decode( encoding='ISO-8859-1')
            body = {"action":"install", "Content-Type": "multipart/form-data", "dangerous": True, "snap": bytes_string,"filename": snap_file_name, "system-restart-immediate": True}
            req = self.session.post("http://snapd/v2/snaps", json = json.loads(json.dumps(body)))
            print(req.json())
            return True
        except Exception as e:
            print(e)
            return False


def main(args=None):
    RefreshSnap_node = RefreshSnap()

if __name__ == '__main__':
    main()

But this never works and it outputs

{‘type’: ‘error’, ‘status-code’: 400, ‘status’: ‘Bad Request’, ‘result’: {‘message’: ‘cannot install: no install/refresh information results from the store’}}

Does anyone know how i can make it work?

2

Hi Rahul, as described here, snaps can be sideloaded with a multipart/form-data request. This is not a json buffer as in many other REST API requests.

As an example, the following code snippet for intall_snap make use of MultipartEncoder to create the data and the headers for the request.

from requests_toolbelt import MultipartEncoder

    def intall_snap(self):
        try:
            snap_file_name = "turftank-mk1-sw_2.1.2_arm64.snap"

            mp_encoder = MultipartEncoder(fields = {"action":"install", "dangerous": "true", "snap": (snap_file_name, open(snap_file_name,'rb')), "system-restart-immediate": "true"})

            head ={"Content-Type": mp_encoder.content_type}
#            print(head)
#            print(mp_encoder.to_string())
            req = self.session.post("http://snapd/v2/snaps", data=mp_encoder, headers=head)

            print(req.json())
            return True
        except Exception as e:
            print(e)
            return False
3

Hi, Now here if i want to validate the “turftank-mk1-sw_2.1.2_arm64.assert” and then install install the snap . how would i do that?

Is it through POST /v2/assertions end point?

4

That’s correct. Consider that, in case of a refresh (or install then removal, then install again), the assertions has been already added on the device so potentially you don’t need add them again. In any case I recommend to re-add the assertions when sideloading in case some of those are changed in the meantime (e.g. snap-declaration changed because an autoconnection interface has been added)

5

So, i wouldnt have to install it as dangerous right?

Also, Is it possible to provide an example for post assertion request?

6

Correct.

Below intall_snap modified:

    def intall_snap(self):
        try:
            snap_file_name = "test-diegobruno_4.snap"
            assert_file_name = "test-diegobruno_4.assert"
            with open(assert_file_name,'rb') as f :
               file_data = f.read()

            mp_encoder = MultipartEncoder(fields = {"action":"install", "snap": (snap_file_name, open(snap_file_name,'rb')), "system-restart-immediate": "true"}) #dangerous removed

            head ={"Content-Type": mp_encoder.content_type}
#           print(head)
#           print(mp_encoder.to_string())

            req = self.session.post("http://snapd/v2/assertions", file_data)
            print(req.json())

            req = self.session.post("http://snapd/v2/snaps", data=mp_encoder, headers=head)
            print(req.json())

            return True
        except Exception as e:
            print(e)
            return False
1 Like
7

Hi @dbruno74,

How would the install endpoint work when installing multiple snaps at the same time?

Would we have to update snaps one by one or can we install multiple snaps at the same time?

8

The install endpoint accepts the installation of multiple snaps, just add others “snap” parts like in the example below:

    def intall_snap(self):
        try:
            snap_file_name = "test-diegobruno_4.snap"
            snap_file_name_2 = "tcpdump-snap_25.snap"
            with open(assert_file_name,'rb') as f :
               file_data = f.read()

            mp_encoder = MultipartEncoder([("action", "install"), ("snap", (snap_file_name, open(snap_file_name,'rb'))), ("snap", (snap_file_name_2, open(snap_file_name_2,'rb')))])
            head ={"Content-Type": mp_encoder.content_type}
            print(head)
#           print(mp_encoder.to_string())

            req = self.session.post("http://snapd/v2/snaps", data=mp_encoder, headers=head)
            print(req.json())

            return True
        except Exception as e:
            print(e)
            return False

Note that the parts are expressed with a list instead of a dictionary, this is needed to be able to add parts with the same name (“snap”)

1 Like