gsm_poc_token.py

#!/usr/bin/env python3

import telnetlib
import sqlite3
import sys
import os
import hmac, hashlib
import requests


TOKEN_BASE_URL="http://someurltothepoc.de/phonebook/?token=%s&hash=%s"
OPENBSC_HOST="127.0.0.1"
OPENBSC_PORT=4242
HMAC_KEY = b'abcdefghijklmnopqrstuvwxyzabcdef'

import time

class OsmoNITB(object):

    PROMPT = 'OsmoHLR#'
    
    def __init__(self, host, port):
        self.lnk = telnetlib.Telnet(host, port)
        self.command("enable");
        self.check_prompt()

    def command(self, cmd):
        self.lnk.write(cmd + "\n")

    def check_prompt(self):
        v = self.lnk.read_until(self.PROMPT, 1)
        if not v.endswith(self.PROMPT):
            raise RuntimeError("Invalid prompt")
        return v[:-len(self.PROMPT)]

    def find_info_for_ext(self, ext):
        self.command("show subscriber extension %s" % ext)
        r = self.check_prompt()

        if 'No subscriber found for extension' in r:
            return None

        print(r)

        d = dict([
            [y.strip() for y in x.strip().split(':',1)]
            for x in r.splitlines()[1:]
        ])

        # Manual fix for ID
        d['Authorized'] = d['ID'].split(',')[1].split(':')[1].strip()
        d['ID'] = d['ID'].split(',')[0]

        return d

    def free_extension(self, ext, info=None):
        # Get info if needed
        if info is None:
            info = self.find_info_for_ext(ext)

        # No info -> extension is free
        if info is None:
            return

        # Reassign extension to default one
        self.command("subscriber id %s extension 99%s" % (info['ID'], info['ID']))
        self.check_prompt()

    def update_extension(self, imsi, ext, name):
        self.command("subscriber imsi %s extension %s" % (imsi, ext))
        self.check_prompt()
        self.command("subscriber imsi %s name %s" % (imsi, name))
        self.check_prompt()

    def bust_cache_for_ext(self, ext):
        self.command("subscriber extension %s update" % ext)
        self.check_prompt()

class InvalidToken(RuntimeError):
    pass
class WrongToken(RuntimeError):
    pass

def check_token(token):
    btoken = bytes(token, 'ascii')

    if len(token) != 11 or token[0] == '0':
        raise InvalidToken

    # Check validity & get extension
    token_hmac = hmac.new(HMAC_KEY, btoken, hashlib.sha256).hexdigest()
    resp = requests.get(TOKEN_BASE_URL % (token, token_hmac))
    answer = str(resp.content, 'utf-8')
    answer = answer.strip().split(" ")
    if len(answer) != 3:
        raise WrongToken
    return answer

def main():
    if len(sys.argv) != 3:
        raise RuntimeError("ext_assign_hmac.py old_msisdn new_msisdn")

    old_msisdn = sys.argv[1]
    token = sys.argv[2]

    answer = check_token(token)

    dummy, extension, name = answer

    # Get OsmoNITB link
    lnk = OsmoNITB(OPENBSC_HOST, OPENBSC_PORT)

    # Get info for that extensions
    lnk.free_extension(extension)

    # Update with new data
    lnk.update_extension(old_msisdn, extension, name)

if __name__ == '__main__':
    main()