Source code for bigbluebutton.api.bigbluebutton

"""Data structures to handle BigBlueButton servers.

This module also contains group handling support and the implementation
of the communication with the API itself. Meetings and attendees manage
themselves by talking to the :class:`BigBlueButton` object linked to them. All
other methods should be called on a :class:`BigBlueButtonGroup`, which takes care
of routing to the appropriate server(s).

import concurrent.futures
import json
import logging
import random
import subprocess  # noqa: S404
from dataclasses import dataclass, field
from hashlib import sha1
from socket import getfqdn
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
from urllib.parse import urlencode
from uuid import uuid1

import requests
import xmltodict

from ._caching import cache
from .attendee import Attendee
from .loadbalancing import CHECKERS as lb_checkers
from .meeting import Meeting

    from sadf import SadfCommand, SadfReport
    import sadf.fieldgroups as sadf_fieldgroups
except ImportError:  # pragma: no cover
    SadfCommand, SadfReport, sadf_fieldgroups = None, None, None

logger = logging.getLogger(__name__)

[docs]@dataclass class BigBlueButton: """One BigBlueButton server. A BigBlueButton instance holds the information needed to communicate with the BigBlueButton server. When creating meetings, it passes itself to the :class:`~bigbluebutton.api.meeting.Meeting` object, and provides methods for the Meeting object to manage itself. A BigBlueButton server always belongs to a :class:`BigBlueButtonGroup`, which it gets passed in the constructor and it registers itself with. :param group: Reference to the :class:`BigBlueButtonGroup` this server belongs to :param name: Descriptive name of this server :param url: Full URL to the API endpoint of this BigBlueButton server (with trailing /) :param salt: API secret, as reported by bbb-conf on the server :param host: Hostname of the server, e.g. for SSH access (guessed from URL if unset) :param request_timeout: Timeout for API requests, either one value or a two-tuple defining TCP and HTTP request timeouts """ group: "BigBlueButtonGroup" name: str url: str salt: str host: Optional[str] = field(default=None, compare=False) meetings: Dict[str, "Meeting"] = field(default_factory=dict, init=False, compare=False) sysstat: Optional[SadfReport] = field(default=None, init=False, compare=False) request_timeout: Union[float, Tuple[float, float]] = field(default=(0.5, 10), compare=False) cache_timeout: float = field(default=30, compare=False) def __post_init__(self): """Set up the server object to work properly. 1. Self-register with the :class:`BigBlueButtonGroup` passed. 2. Set-up HTTP session store (for cookies, etc.). """[] = self logger.debug(f"Self-registered server {} in group {}") # Use persistent HTTP session to track JSESSIONID cookie (et al) self._session = cache.get(f"{self.url}::session") or requests.Session() # Restore meeting and sysstat data from last time if any self.meetings = cache.get(f"{self.url}::meetings") or {} self.sysstat = cache.get(f"{self.url}::sysstat") def __del__(self): """Store some data for next time.""" cache.set(f"{self.url}::meetings", self.meetings, self.cache_timeout) cache.set(f"{self.url}::session", self._session, self.cache_timeout) cache.set(f"{self.url}::sysstat", self.sysstat, self.cache_timeout)
[docs] @staticmethod def request_checksum(call: str, query: str, salt: str) -> str: """Compute the checksum needed to authenticate API requests to a BigBlueButton server. The checksum has to be sent with every API request and is constructed like this: 1. Build the full query string (already done when passed into this method) 2. Prepend the name of the API call, without delimiter 3. Append the API salt (shared secret, key,…) provided by the BBB server 4. Calculate the SHA1 sum of the resulting string >>> BigBlueButton.request_checksum("isMeetingRunning", "meetingID=Foo", "MyTestSalt") 'f59b73c5cf1db387da4ca7d937049420d4c50a12' The resulting checksum is added tothe original query string as a parameter called checksum. :param call: Name of the API method to call (last part of URL) :param query: urlencoded query string with call parameters :param salt: The API secret the server expects requests to be signed with :return: The checksum (to be appended as the checksum parameter) """ hash_string = call + query + salt checksum = sha1(hash_string.encode()).hexdigest() # noqa: S303, we have no choice return checksum
def _build_url(self, call: str, params: Optional[Dict[str, str]] = None) -> str: # Generate query string with challenge-response checksum # cf. query = urlencode(params or {}) query += "&checksum=" + self.request_checksum(call, query, self.salt) url = f"{self.url}{call}?{query}" return url def _request(self, call: str, params: Optional[Dict[str, str]] = None) -> Dict[str, Any]: url = self._build_url(call, params) res = self._session.get(url, timeout=self.request_timeout) xml = xmltodict.parse(res.text) return xml["response"]
[docs] def create_meeting(self, do_create: bool = True, *args, **kwargs) -> Meeting: """Create a meeting on thie BigBlueButton server. args and kwargs are passed verbatim into the :class:`~bigbluebutton.api.meeting.Meeting` constructor. The additional parameter do_create determines whether the API is actually called to create the meeting. If it is False, tje object is returned but the meeting is not transferred to the BigBlueButton server. This method first looks up whether a meeting with the passed meeting_id and reuse that object if a meeting is found. If none is found, a new object is created and the create API call sent to the server (if do_create is not set to False). The BigBlueButton API guarantees that the create call is idempotent, so calling this method on a meeting ID of an existing meeting is safe. But, as every server is part of a server group, consumers should always call create_meeting on a :class:`BigBlueButtonGroup` instead (or use its meeting property to make use of caching). :param do_create: True to call the create API method on the server, False to only create the Meeting object :return: The found or newly created Meeting object """ if "meeting_id" in kwargs and kwargs["meeting_id"] in self.meetings: meeting = self.meetings[kwargs["meeting_id"]] else: meeting = Meeting(self, *args, **kwargs) if do_create: meeting.create() else: meeting.get_meeting_info() return meeting
[docs] def get_meetings(self) -> Dict[str, "Meeting"]: """Get all meetings known on the BigBlueButton server. This method calls the getMeetings API call on the BigBlueButton server and constructs Meeting objects from all meetings in the result. For meetings that are already known as objects, the appropriate object is returned; for all unknown meetings, new objects are created. As every server is part of a server group, consumers should always call :func:`BigBlueButtonGroup.create_meeting` instead. :return: A dictionary mapping meeting IDs to Meeting objects """"Updating meetings on server {}") res = self._request("getMeetings") found_meeting_ids = [] if "meetings" not in res or not res["meetings"] or not res["meetings"]["meeting"]: self.meetings.clear()"Cleared all meetings from server {}") else: if not isinstance(res["meetings"]["meeting"], list): res["meetings"]["meeting"] = [res["meetings"]["meeting"]] for meeting_dict in res["meetings"]["meeting"]: meeting_id = meeting_dict["meetingID"] if meeting_id in self.meetings: meeting = self.meetings[meeting_id] logger.debug(f"Meeting {meeting_id} already known on server {}") else: meeting = Meeting(self, meeting_id=meeting_id) self.meetings[meeting_id] = meeting logger.debug(f"Meeting {meeting_id} discovered, adding to server {}") meeting._update_from_response(meeting_dict) # Track found IDs for later clean up found_meeting_ids.append(meeting_id) # Clean up meetings not known anymore for meeting_id in list(self.meetings.keys()): if meeting_id not in found_meeting_ids: del self.meetings[meeting_id] return self.meetings
[docs] def ssh_command( self, command: Sequence[str], input_: Optional[str] = None ) -> subprocess.CompletedProcess: """Execute a shell command through an SSH connection to the server. This method constructs an ssh command and runs it as a subprocess, without shell expansion. However, shell expansion WILL take place on the server side. The subprocess result is returned verbatim. :param command: Sequence of command words, like in argv :param input_: Text to pipe into the command on stdin :return: The verbatim subprocess object (including stdout, stderr, and return code) """ if is None: raise ValueError(f"Server {} has no hostname set") args: List[str] = ["ssh",] + list(command)"Running SSH command {args} on {}") res = # noqa: S603, command injection is intended args, capture_output=True, input=input_, text=True ) return res
[docs] def get_sysstat(self) -> Optional[SadfReport]: """Get the output of the sadf/sar command (from the sysstat package) through SSH. The result is used when load-balancing API requests. To use this functaionality, the sadf Python package (in the "sysstat" extra) must be installed. """ if SadfCommand is None: logger.warn(f"sysstat requested on host {}, but python-sadf unavailable") self.sysstat = None return None"Getting system statistics on server {}") # Let the python-sadf library build a command, but do not run it sadf_cmd = SadfCommand() sadf_cmd.field_groups = [ sadf_fieldgroups.CPULoad(all_fields=True), sadf_fieldgroups.IO(), sadf_fieldgroups.Kernel(), sadf_fieldgroups.Memory(all_fields=True), sadf_fieldgroups.Network(dev=True, edev=True, sock=True), sadf_fieldgroups.Queue(), ] cmd = sadf_cmd._build_command() # Prepend environment python-sadf woud use cmd = ["env"] + [f"{k}={v}" for k, v in sadf_cmd._command_env.items()] + cmd # Execute using our own runner and extract data ret = self.ssh_command(cmd) if ret.returncode == 0: try: host_data = json.loads(ret.stdout)["sysstat"]["hosts"][0] except (KeyError, json.JSONDecodeError): logger.error(f"sadf produced invalid data on host {}") else: # Hand back to python-sadf to generate pandas report self.sysstat = SadfReport(host_data, sadf_cmd.field_groups) else: logger.warn(f"sadf unavailable or failing on host {}") return self.sysstat
[docs] def refresh(self) -> None: """Refresh various aspects of this server. - Meetings - Sysstat data """ logger.debug(f"Fully refreshing server {}") self.get_meetings() self.get_sysstat()
[docs]@dataclass class BigBlueButtonGroup: """A set of BigBlueButton servers. All API operations should be called on this object, instead of a single server. It takes care of aggregating results from the backend servers (e.g. for getMeetings) and of routing requests to the correct server (e.g. for loadbalancing on create). Servers (:class:`BigBlueButton` objects) register themselves to the group passed as first argument on instantiation. If only one server has to be handled, still place it in a group. :param name: Descriptive name of this server group :param workers: Number of servers to operate on parallely, e.g. when retrieving meetings or running SSH commands) :param logout_url: Default URL to redirect clients to on meeting end :param origin: Name of the origin software passed as meeting meta-data :param origin_server_name: Hostname of the origin, passed as meeting meta-data; defaults to the systems's local FQDN :param generate_meeting_id_cb: Callable to run for generating meeting IDs """ name: str apis: Dict[str, BigBlueButton] = field(default_factory=dict, init=False, compare=False) workers: int = field(default=10, compare=False) logout_url: Optional[str] = field(default=None, compare=False) origin: str = "python-bigbluebutton2" origin_server_name: str = field(default_factory=getfqdn) generate_meeting_id_cb: Optional[Callable[[], str]] = field(default=None, compare=False)
[docs] def generate_meeting_id(self) -> str: """Generate a unique meeting ID. By default, this returns a GUID, unless a callback was defined in the `generate_meeting_id_cb` attribute. """ if self.generate_meeting_id_cb is None: return str(uuid1()) else: return self.generate_meeting_id_cb()
@property def meetings(self) -> Dict[str, "Meeting"]: """All meetings known on all servers in the group. Each meeting will reference the backend server it is actually running on, so calling methods on the Meeting objects returned in this property always does "the right thing". """ res = {} for name, api in self.apis.items(): res.update(api.meetings) return res
[docs] def new(self, name: str, *args, **kwargs) -> BigBlueButton: """Createa new :class:`BigBlueButton` server object in this group. The server object will register itself in this group. For the arguments, see the documentation of :class:`BigBlueButton`. """ logger.debug(f"Creating new API client {name}") bbb = BigBlueButton(self, name, *args, **kwargs) return bbb
def _foreach(self, method: str, *args, **kwargs) -> Dict[str, Any]: logger.debug( f"Calling method {method} on all servers in group {} ({self.workers} workers)" ) res = {} with concurrent.futures.ThreadPoolExecutor(max_workers=self.workers) as pool: futures = {} for name, bbb in self.apis.items(): fn = getattr(bbb, method) futures[pool.submit(fn, *args, **kwargs)] = name logger.debug(f"Pooled method {method} on server {name}") logger.debug("Waiting for pooled methods") for future in concurrent.futures.as_completed(futures): name = futures[future] res[name] = future.result() logger.debug(f"Pooled method {method} on server {name} returned") return res
[docs] def get_meetings(self) -> Dict[str, "Meeting"]: """Update meeting information on all server objects. It is aggregated into the meetings attribute. If forcefully updating the metting information cache is not intended, the meeting property should be used instead. """ self._foreach("get_meetings") return self.meetings
[docs] def ssh_command( self, command: Sequence[str], input_: Optional[str] = None ) -> Dict[str, subprocess.CompletedProcess]: """Call an SSH command on each backend server in this group. For the arguments and behaviour, see the documentation of :func:`BigBlueButton.ssh_command`. The return value is a dictionary with the server name as key and the result of the server's ssh_command method as value. """ res = self._foreach("ssh_command", command, input_) return res
[docs] def get_sysstat(self) -> Dict[str, Optional[SadfReport]]: """Get sysstat information from all backend servers. For the arguments and behaviour, see the documentation of :func:`BigBlueButton.get_sysstat`. The result is a dictionary with the server name as key and the result of the server's get_sysstat method as values. """ res = self._foreach("get_sysstat") return res
[docs] def refresh(self) -> None: """Refresh all backend servers. For a description of the behaviour, see the documentation of :func:`BigBlueButton.refresh`. """ self._foreach("refresh")
[docs] def select_api(self, criteria: Optional[Dict[str, Any]] = None) -> BigBlueButton: """Select a target server in the group by some creiteria. This method runs a ranking algorithm on all servers in the group, then returns the server with the highest ranking. This can be used for load-balancing and similar tasks. The ranking algorithm works like this: 1. Start with giving each server a ranking of N points. 2. Apply a list of checkers on each server. Each checker returns a factor by which to multiply the last known ranking (between 0 and 1). 3. Return the server with the highest ranking. If multiple servers share the same highest ranking, fall back to round robin. A ranking of 0 means a server is unavailable. Servers with a ranking of 0 are never returned. :param criteria: A dictionary of criteria taken into account by each checker. See the documentation of the checkers in the :module:`bigbluebutton.api.loadbalancing` module for details on what criteria exist and how they influence server selection. :return: One BigBlueButton server that best matches the criteria """ # Start with a ranking of 10 for each API apis = {api_name: 10.0 for api_name in self.apis} # Define checkers to run on each API # Each checker returns a factor to scale the ranking by checkers = lb_checkers # Apply checkers, in order for checker in checkers: for api_name, ranking in apis.items(): factor = checker(self.apis[api_name], criteria) apis[api_name] *= factor # Get maximum ranking and all APIs that got it max_ranking = max(apis.values()) apis_won = [ self.apis[api_name] for api_name, ranking in apis.items() if ranking == max_ranking ] # Select random API from winners api = random.choice(apis_won) # noqa: S311, not cryptographic return api
[docs] def create_meeting(self, do_create: bool = True, *args, **kwargs) -> Meeting: """Create a new meeting on one of the backend servers. For the arguments, see the documentation on :class:`~bigbluebutton.api.meeting.Meeting`. This method first tries to find an existing meeting. The original create API method in BigBlueButton is idempotent, so we must ensure we also fulfill this requirement when handling several servers. If this group already has a list of known meetings, and the meeting is found in it, the call is redirected to the backend server it is linked to. If it is not found, :func:`get_meetings` is called to refresh the cache to be certain. If all attempts to find a meeting with the ID fail, or no ID is passed, a backend server is selected using :func:`select_api` and the create call routed there. """ if "meeting_id" in kwargs: meeting = self._find_meeting(kwargs["meeting_id"]) else: meeting = None if meeting: api = meeting.api"Found meeting with id {meeting.meeting_id} on server {}") else:"Creating new meeting on one server in group {}") api = self.select_api() return api.create_meeting(do_create, *args, **kwargs)
def _find_meeting(self, meeting_id: str) -> Optional[Meeting]: if meeting_id in self.meetings or meeting_id in self.get_meetings(): return self.meetings[meeting_id] return None
[docs] def is_meeting_running(self, meeting_id: str) -> bool: """Determine whether a meeting with a given ID is running on any backend server.""" meeting = self._find_meeting(meeting_id) if meeting: return meeting.is_meeting_running() else: return False
[docs] def end_meeting(self, meeting_id: str, password: Optional[str] = None) -> None: """End a meeting determined from the passed ID. If the meeting is not found, this method silently does nothing. If a password is passed as second argument, it is compared against the moderarotr password of the meeting before the end request is even sent to the backend. """ meeting = self._find_meeting(meeting_id) if meeting: # Verify password before even sending the call, if provided # Entirely useless because anyone who can call end can also call getMeetingInfo, # but let's play along… if password is not None and password != meeting.moderator_pw: raise ValueError("The supplied moderatorpassword does not match.") meeting.end()
[docs] def handle_from_data( self, method: str, attrs: Optional[dict] = None, content: Optional[str] = None, filter_meta: Optional[dict] = None, ) -> Dict[str, Any]: """Handle an API call from a method, a dict of parameters, and a text body. This method is useful to pass on structures like HTTP requests on to the library, e.g. when writing a proxy or load balancer. The method name and parameter dictionary are expected to match the official BigBlueButton API. The return value is a dictionary representation of the XML document as would be returned by a BigBlueButton server when called with the same parameters. """ if not attrs: attrs = {} meeting: Optional["Meeting"] if method == "create": kwargs = Meeting.get_kwargs_from_url_args(attrs) meeting = self.create_meeting(**kwargs) return meeting.to_dict() elif method == "join": meeting = self._find_meeting(attrs["meetingID"]) if meeting: kwargs = Attendee.get_kwargs_from_url_args(attrs, meeting) attendee = Attendee(meeting=meeting, **kwargs) attendee.join() # We need to construct this one response manually, because for some inobvious # reason the designers of the BBB API changed their minds and started using # snake_case instead of dromedarCase return { "meeting_id": meeting.meeting_id, "user_id": attendee.user_id, "auth_token": attendee.auth_token, "session_token": attendee.session_token, "url": attendee.url, } else: raise KeyError("Meeting not found.") elif method == "isMeetingRunning": running = self.is_meeting_running(attrs["meetingID"]) return {"running": "true" if running else "false"} elif method == "end": self.end_meeting(attrs["meetingID"], attrs["password"]) return {} elif method == "getMeetingInfo": meeting = self._find_meeting(attrs["meetingID"]) if meeting: meeting.get_meeting_info() return meeting.to_dict() else: raise KeyError("Meeting not found.") elif method == "getMeetings": meetings = self.get_meetings() return { "meetings": [ {"meeting": meeting.to_dict()} for meeting_id, meeting in meetings.items() ] } elif method == "getRecordings": raise NotImplementedError(f"Method {method} not implemented yet") elif method == "publishRecordings": raise NotImplementedError(f"Method {method} not implemented yet") elif method == "deleteRecordings": raise NotImplementedError(f"Method {method} not implemented yet") elif method == "updateRecordings": raise NotImplementedError(f"Method {method} not implemented yet") elif method == "getDefaultConfigXML": raise NotImplementedError(f"Method {method} not implemented yet") elif method == "setConfigXML": raise NotImplementedError(f"Method {method} not implemented yet") elif method == "getRecordingTextTracks": raise NotImplementedError(f"Method {method} not implemented yet") elif method == "putRecordingTextTrack": raise NotImplementedError(f"Method {method} not implemented yet") else: raise TypeError(f"Method {method} is unknown")